👩‍💻 Celery + Redis/RabbitMQ — таски и очереди без боли

Если у тебя Python-сервис начинает «тормозить» из-за тяжёлых операций (отправка почты, обработка файлов, запросы к API) — не надо делать time.sleep(999). Всё это можно вынести в фон через Celery.

Celery = таски + брокер сообщений (Redis или RabbitMQ) + воркеры, которые жрут эти таски.

📦 Установка:

pip install celery[redis]
pip install redis
# или если брокер RabbitMQ:
# pip install celery[rabbitmq]

🧠 Минимальный пример — таска с Redis:

# tasks.py
from celery import Celery

app = Celery(
    "my_app",
    broker="redis://localhost:6379/0",     # Redis
    backend="redis://localhost:6379/0"     # для хранения результата
)

@app.task
def add(x, y):
    return x + y

Запускаем воркера:

celery -A tasks worker --loglevel=info

В другом файле:

from tasks import add
result = add.delay(4, 6)   # отправляем в очередь
print(result.get())        # ждем выполнения → 10

➡️ Всё: воркер получил таску, посчитал, вернул результат.

📛 RabbitMQ вместо Redis
RabbitMQ — более надёжный брокер, с очередями и персистентностью.

Просто меняем конфиг:

app = Celery("my_app", broker="pyamqp://guest@localhost//")

➡️ Дальше код остаётся тем же.


🔒 Ретраи и устойчивость

Celery умеет сам повторять таску при сбое:

@app.task(bind=True, max_retries=3, default_retry_delay=5)
def fragile_task(self, url):
    try:
        # какая-то логика
        return "ok"
    except Exception as exc:
        raise self.retry(exc=exc)  # автоматический retry

➡️ Если упало — повторим через 5 секунд, максимум 3 раза.

📂 Планировщик задач (Celery Beat)

Можно запускать таски по расписанию:

from celery.schedules import crontab

app.conf.beat_schedule = {
    "clear-cache-every-night": {
        "task": "tasks.clear_cache",
        "schedule": crontab(hour=3, minute=0),  # каждый день в 03:00
    },
}

Запускаем:

celery -A tasks beat

➡️ Всё, у тебя свой cron прямо в Celery.

🧱 Масштабирование

🟢 Redis — проще, быстрее поднять, но хуже с очень тяжёлой нагрузкой.
🟢 RabbitMQ — надёжнее, умнее с очередями (ack, приоритеты, персистентность).
🟢 Celery умеет параллельные воркеры, пулы процессов/потоков.

🎯 Полезные фишки:

🟢 .delay() — отправка таски асинхронно

🟢 .apply_async(countdown=10) — отложить на 10 секунд

🟢 .get(timeout=5) — дождаться результата с таймаутом

🟢 Результаты можно хранить не только в Redis, но и в SQL, Mongo, S3

🗣️ Запомни: Celery — это «рабочие руки» для твоего кода.Ты пишешь только задачу, а кто, где и когда её выполнит — решают воркеры и брокер.

🚀 SQL + ORM: от ручного SQL к магии Python

SQL — это язык запросов к базе.
ORM (Object–Relational Mapping) — это прослойка: вместо "писать руками SQL" ты работаешь с объектами.
Python делает это через SQLAlchemy или Django ORM.

📊 Шаг 1 — без ORM, чистый SQL

import sqlite3

conn = sqlite3.connect("app.db")
cur = conn.cursor()

cur.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
cur.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
conn.commit()

cur.execute("SELECT * FROM users")
print(cur.fetchall())

➡️ Всё честно: таблица, insert, select. Но руками писать SQL = боль и дублирование.

🧩 Шаг 2 — ORM превращает таблицу в класс

SQLAlchemy (пример на SQLite):

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, Session

engine = create_engine("sqlite:///app.db")
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)

Base.metadata.create_all(engine)

# работа через объекты
with Session(engine) as session:
    session.add(User(name="Alice"))
    session.commit()

    users = session.query(User).all()
    print([u.name for u in users])

➡️ Таблица = класс. Строка = объект. Запрос = метод ORM.

🛠 Шаг 3 — Django ORM (уже встроен в фреймворк)

models.py в Django-проекте:

from django.db import models

class User(models.Model):
    name = models.CharField(max_length=100)

➡️ Django сам создаст таблицу app_user.

Создание и запросы:

User.objects.create(name="Alice")
users = User.objects.all()
print([u.name for u in users])

➡️ Всё без SQL — только объекты и методы.

📦 Шаг 4 — фильтрация и условия

SQLAlchemy:

users = session.query(User).filter(User.name == "Alice").all()

Django ORM:

users = User.objects.filter(name="Alice")

➡️ В коде ты пишешь Python, под капотом генерируется SQL:
SELECT * FROM users WHERE name='Alice'

🔗 Шаг 5 — связи (one-to-many)

SQLAlchemy:

from sqlalchemy.orm import relationship, Mapped, mapped_column, DeclarativeBase

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="posts")

User.posts = relationship("Post", back_populates="user")

➡️ Теперь user.posts даст список постов, а post.user вернёт автора.

Django:

class Post(models.Model):
    title = models.CharField(max_length=100)
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="posts")

➡️ user.posts.all() и post.user работают как магия.

🧪 Шаг 6 — миграции

🟢 SQLAlchemy: через Alembic (alembic revision --autogenerate)
🟢 Django: встроенные миграции (python manage.py makemigrations && migrate)

➡️ Ты меняешь класс → ORM меняет схему в базе.


🗣️ Запомни:ORM не отменяет знание SQL: фильтры, JOIN-ы и транзакции остаются под капотом.

🔠 Type Hints: код говорит за тебя

Python — динамический, можешь писать без типов.
Но с аннотациями : int, : str код начинает сам себя документировать.
Типы не проверяются в рантайме — но помогают читать, дебажить и ловить ошибки раньше.

📛 Пример: аннотации в функции

def add(a: int, b: int) -> int:
    return a + b

➡️ В рантайме работает так же.
Но теперь IDE и mypy понимают: на входе и выходе только числа.

🧪 Комбинации типов

from typing import List, Dict

def users_to_dict(users: List[str]) -> Dict[str, int]:
    return {u: len(u) for u in users}

➡️ Python по-прежнему не строгий.
Но твой код стал декларативным: видно, что ждёт функция.

📦 Опциональные значения

from typing import Optional

def find_user(id: int) -> Optional[str]:
    if id == 0:
        return None
    return "admin"

➡️ Без if value is None не разберёшься.
А с Optional всё ясно из сигнатуры.

🔐 Собственные типы

from typing import NewType

UserId = NewType("UserId", int)

def get_user(id: UserId) -> str:
    return f"user-{id}"

➡️ UserId и int одинаковы в рантайме,
но для анализатора — это разные сущности.

🪄 Магия PEP 604 (Python 3.10+)

def foo(x: int | str) -> None:
    ...

➡️ Больше не нужно Union.
Теперь типы можно писать через |.


🗣️ Запомни: type hints — это декларация смысла, а не ограничение.

👩‍💻 Цикл while, Операторы break и continue

В этом видео автор подробно объясняет, как работает цикл while в Python.
Разбираются операторы break и continue, показано, как с их помощью управлять выполнением цикла. Всё демонстрируется на простых примерах, что помогает легко понять и применить эти конструкции на практике.

➡️ Ссылка на первоисточник

🤩 Pytstart || #Видеокурс

🖥 Как работает zip() в Python: синхронный перебор

🧵 zip() объединяет несколько итерируемых объектов в кортежи по элементам.

🔧 Пример:

names = ["Alice", "Bob", "Charlie"]
scores = [85, 92, 78]

for name, score in zip(names, scores):
    print(f"{name}: {score}")

📌 Вывод:

Alice: 85  
Bob: 92  
Charlie: 78

❗️ Что важно знать

- zip останавливается по самому короткому из входов.
- Можно “распаковать” с помощью zip(*iterables).

pairs = [("a", 1), ("b", 2)]
letters, numbers = zip(*pairs)
print(letters)  # ('a', 'b')

🗣 Используется в парном переборе, обработке CSV, генерации пар "ключ-значение", и везде, где нужно синхронизировать данные.

🎭 Decorators в Python: меняем поведение функции без переписывания

Ты написал:

def greet(name):
    print(f"Привет, {name}!")

А потом захотел добавить логирование, проверку ошибок или права доступа. Переписывать код не хочется.

👩‍💻 Python тебе ничего не скажет. Но выход есть — декораторы. Это «обёртки» вокруг функций, которые добавляют новое поведение.


📦 Пример: простое логирование

def log_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"Вызов {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

@log_decorator
def greet(name):
    print(f"Привет, {name}!")

greet("Анна")

➡️ Функция greet осталась чистой, логика вынесена наружу.

🔑 Декораторы с параметрами

def repeat(times):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for _ in range(times):
                func(*args, **kwargs)
        return wrapper
    return decorator

@repeat(3)
def hello():
    print("👋 Привет")

hello()

➡️ Теперь hello() вызывается 3 раза подряд.

⚠️ Без functools.wraps теряется имя функции

from functools import wraps

def safe(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            print("Ошибка:", e)
    return wrapper

@safe
def divide(a, b):
    return a / b

➡️ @wraps сохраняет имя и docstring, иначе функция станет просто wrapper.


🧪 Практика: где декораторы реально нужны

👍 Проверка прав

def require_admin(func):
    def wrapper(user, *args, **kwargs):
        if not user.get("is_admin"):
            raise PermissionError("⛔️ Нет доступа")
        return func(user, *args, **kwargs)
    return wrapper

@require_admin
def delete_user(user, uid):
    print(f"Удалён {uid}")

👍 Обработка ошибок

@safe
def risky(x):
    return 10 / x

risky(0)  # Ошибка перехвачена

👍 Комбинация декораторов

@safe
@log_decorator
def risky_op(x):
    return 10 / x

➡️ Сначала лог, потом защита. Порядок имеет значение.

🗣️ Запомни: Декоратор = функция, которая принимает и возвращает функцию.

🔍 __str__🆚__repr__ : зачем два разных отображения объекта

В Python у любого объекта можно определить два разных способа отображения — __str__ и __repr__. Кажется избыточным, но на деле это основа читаемости и отладки.

Разберём всё по полочкам 👇

🔍 str — человекочитаемый вывод

class User:
    def __init__(self, name):
        self.name = name
    def __str__(self):
        return f"Пользователь {self.name}"

u = User("Иван")
print(u)   # Пользователь Иван

➡️ __str__ нужен для красивого вывода в консоли, логах или интерфейсе. Его цель — понятность.

🛠 repr — точный «отладочный» вывод

class User:
    def __init__(self, name):
        self.name = name
    def __repr__(self):
        return f"User(name={self.name!r})"

u = User("Иван")
print(repr(u))  # User(name='Иван')

➡️ __repr__ всегда должен быть однозначным. Его читают программисты и дебаггер, а не пользователи.

🖼 В интерактивной консоли вызывается repr

u = User("Иван")
u

➡️ Jupyter, IPython и REPL используют __repr__. Поэтому именно он важен для отладки.

📑 Логи: str для людей, repr для разработчиков

print(str(u))   # Пользователь Иван
print(repr(u))  # User(name='Иван')

➡️В логах удобно комбинировать оба: человекочитаемый текст + точная структура.

🎭 Если нет str — используется repr

class Item:
    def __repr__(self):
        return "Item(id=42)"

i = Item()
print(i)  # Item(id=42)

➡️ Python подставляет __repr__, если __str__ не определён. Но наоборот — не работает.

⚡️ Совмещённый подход

class Product:
    def __init__(self, name, price):
        self.name, self.price = name, price
    def __repr__(self):
        return f"Product({self.name!r}, {self.price})"
    __str__ = __repr__   # одинаковый вывод

➡️Иногда удобно оставить один метод и приравнять их. Но чаще нужны оба.

🧩 Лучшие практики

🟢__repr__ должен быть «однозначным» (по возможности — похож на конструктор).
🟢 __str__ должен быть «дружелюбным» — читаться как текст.
🟢Если сомневаешься — реализуй хотя бы __repr__.

🗣️ Запомни:str → для пользователя, repr → для разработчика.В консоли, дебаге и коллекциях срабатывает именно repr.Грамотное разделение делает код удобнее и чище, а логи — понятнее.

👩‍💻 Глобальные и локальные переменные в Python — почему UnboundLocalError ломает код

В Python переменные живут в разных областях видимости: локальной (функция), глобальной (модуль), и встроенной (len, print, и т. д.).
Новички часто сталкиваются с UnboundLocalError — когда кажется, что переменная доступна, но интерпретатор думает иначе.

⚡️ Ошибка №1: доступ к глобальной переменной внутри функции

x = 10

def foo():
    print(x)   # 🔴 UnboundLocalError
    x = 5

foo()

➡️ Python видит x = 5 и считает, что x — локальная переменная. Но до присвоения её нет, поэтому ошибка.

🌍 Решение: объявить переменную глобальной

x = 10

def foo():
    global x
    print(x)   # ✅ 10
    x = 5

foo()
print(x)  # 5

➡️ Теперь мы явно сказали: работаем с глобальной x.


🔄 Ошибка №2: изменение замыкания без nonlocal

def outer():
    x = 10
    def inner():
        x = x + 1   # 🔴 UnboundLocalError
        return x
    return inner()

outer()

➡️ Python считает, что x в inner — локальная, а к замыканию он не обращается.

🔧 Решение: использовать nonlocal

def outer():
    x = 10
    def inner():
        nonlocal x
        x = x + 1
        return x
    return inner()

print(outer())  # ✅ 11

➡️ nonlocal говорит: используй переменную из внешней функции.


🧪 Практика: счётчик на замыканиях

def make_counter():
    count = 0
    def inc():
        nonlocal count
        count += 1
        return count
    return inc

counter = make_counter()
print(counter())  # 1
print(counter())  # 2

➡️ Без nonlocal счётчик всегда бы начинался с нуля.


📦 Когда использовать global и nonlocal?

🟢 global — когда нужно изменять переменные уровня модуля (например, конфиг или кеш).
🟢 nonlocal — когда пишешь вложенные функции и нужно работать с переменными из внешней, а не создавать новые.

⚠️ Подводный камень: не меняй глобальные без нужды

users = []

def add_user(name):
    users.append(name)  # ✅ работает без global

➡️ Мы меняем список, а не переназначаем переменную. Поэтому global не нужен.

🗣️ Запомни: UnboundLocalError = Python думает, что переменная локальная. Используй global для изменения глобальных переменных, nonlocal — для замыканий. Избегай глобалок: они делают код запутанным. Лучше возвращать значения и передавать их явно.

🎲 Mutable vs Immutable: список в аргументах функции — неожиданный баг

В Python есть два типа объектов: immutable (строки, числа, кортежи) и mutable (списки, словари, множества).
Разница кажется очевидной, пока не сталкиваешься с функциями и дефолтными аргументами.

⚡️ Классический баг: изменяемый аргумент по умолчанию

def add_item(item, bucket=[]):
    bucket.append(item)
    return bucket

print(add_item("🍎"))  # ['🍎']
print(add_item("🍌"))  # ['🍎', '🍌'] ❌

➡️ Список создаётся один раз при определении функции. Все вызовы делят его между собой.

🔍 Почему так происходит?

Python вычисляет значения аргументов по умолчанию при определении функции, а не при каждом вызове.
То есть bucket=[] создаётся один раз и живёт всё время жизни функции.

✅ Как правильно: использовать None

def add_item(item, bucket=None):
    if bucket is None:
        bucket = []
    bucket.append(item)
    return bucket

print(add_item("🍎"))  # ['🍎']
print(add_item("🍌"))  # ['🍌']

➡️ Теперь список создаётся каждый раз заново, если аргумент не передан.

🧪 Практика: словари ведут себя так же

def register_user(name, db={}):
    db[name] = True
    return db

print(register_user("Alice"))  # {'Alice': True}
print(register_user("Bob"))    # {'Alice': True, 'Bob': True} ❌

✔️ Исправляем через None:

def register_user(name, db=None):
    if db is None:
        db = {}
    db[name] = True
    return db

📦 Когда это реально нужно?

Иногда специально используют «общий объект по умолчанию»:

def cache_query(query, cache={}):
    if query in cache:
        return cache[query]
    result = f"Выполнил {query}"
    cache[query] = result
    return result

➡️ Все вызовы делят одно хранилище. Это работает как встроенный кеш.

🧠 Immutable так не ломаются

def add_one(x=0):
    return x + 1

print(add_one())  # 1
print(add_one())  # 1

➡️ С числами, строками, кортежами проблем нет — они неизменяемые.


🗣 Запомни: Не используй изменяемые объекты в дефолтных аргументах. Безопасный паттерн: arg=None, потом if arg is None: arg = []. Исключение — если тебе нужен общий объект (например, кеш).

🔄 Генераторы и yield: как делать ленивые вычисления и экономить память

Обычные функции в Python возвращают всё сразу. Генераторы — по частям. Это «ленивые» вычисления: результат отдаётся только тогда, когда он нужен. Экономия памяти и скорость работы на больших данных становятся колоссальными.

⚡️ Простейший пример с yield

def numbers():
    yield 1
    yield 2
    yield 3

for n in numbers():
    print(n)

➡️ Функция не возвращает список, а отдаёт значения по одному.

📂 Чтение больших файлов построчно

def read_file(path):
    with open(path, "r") as f:
        for line in f:
            yield line.strip()

for line in read_file("big.log"):
    print(line)

➡️ Файл читается строчка за строчкой, без загрузки всего в память.

♾️ Бесконечные последовательности

def counter(start=0):
    while True:
        yield start
        start += 1

for n in counter():
    if n > 5:
        break
    print(n)

➡️ Генератор может выдавать элементы бесконечно. Список так сделать нельзя.

🔗 Пайплайны генераторов

def numbers():
    for i in range(10):
        yield i

def squared(seq):
    for x in seq:
        yield x * x

for x in squared(numbers()):
    print(x)

➡️ Одни генераторы передают данные другим. Получается конвейер обработки.


🛠 Генераторные выражения

squares = (x*x for x in range(1_000_000))
print(next(squares))  # 1
print(next(squares))  # 4

➡️ Почти как list comprehension, но без хранения всех элементов в памяти.


🟢 Где это реально нужно:

🟢обработка логов и CSV гигабайтного размера;
🟢потоковые данные (сокеты, API);
🟢построение бесконечных последовательностей;
🟢конвейеры обработки (map/filter без лишних списков).

🗣️ Запомни:yield отдаёт результат по частям, не занимая память. Генераторы можно объединять в цепочки — как трубы. Если данных много или они бесконечны — только генераторы спасут.

👩‍💻 Функция def и Возврат return

В этом видео автор объясняет, как создавать функции в Python с помощью ключевого слова def и использовать оператор return для возврата результата. Показано, зачем нужны функции, как они упрощают код и делают его удобным для повторного использования.

➡️ Ссылка на первоисточник

🤩 Pytstart || #Видеокурс

👩‍💻 Модули. Импорт своих модулей

В этом видео разбирается, что такое модули в Python и как их подключать в проект. Автор показывает, как импортировать встроенные модули, а также создавать и использовать свои собственные. Это позволяет делать код более структурированным и переиспользуемым.

➡️ Ссылка на первоисточник

🤩 Pytstart || #Видеокурс

🚦 Python + Redis Streams — быстрая обработка событий

Redis Streams — это структура данных для очередей и событий. С ней можно строить системы, где сообщения не теряются, а обрабатываются быстро и упорядоченно.

📥 Запись события в стрим

import redis
r = redis.Redis()

r.xadd("mystream", {"user": "ivan", "action": "login"})

➡️ Каждое событие получает уникальный ID и хранится в очереди.

📤 Чтение из стрима

events = r.xread({"mystream": "0"}, count=5)
print(events)

➡️ Можно читать все сообщения с начала или только новые.

👥 Консьюмер-группы

r.xgroup_create("mystream", "group1", id="0", mkstream=True)
msg = r.xreadgroup("group1", "consumer-1", {"mystream": ">"}, count=1)

➡️ Несколько воркеров обрабатывают события параллельно, но каждое сообщение достанется только одному.

♻️ Подтверждение обработки

r.xack("mystream", "group1", msg_id)

➡️ Сообщение убирается из «висячих», значит оно точно обработано.

🔄 Повторная доставка

Если воркер упал, Redis отдаст «неподтверждённые» события другому.

➡️ Это делает систему надёжной.

⚡️ Потоковая обработка в реальном времени

while True:
    msgs = r.xread({"mystream": "$"}, block=5000)
    for _, events in msgs:
        for _, data in events:
            print(data)

➡️ Код крутится бесконечно и реагирует на новые события сразу.

🚀 Кейсы применения

Логи, чаты, системы мониторинга, стриминг данных с IoT-устройств. Всё, где важна скорость и порядок.

🗣️ Запомни: Redis Streams позволяют строить быстрые и надёжные пайплайны обработки событий, а с consumer groups ты получаешь масштабируемость и защиту от потерь.

👩‍💻 Pydantic v2 — что изменилось и как теперь валидировать данные

Pydantic v2 — не мелкий апдейт, а рефакторинг движка в отдельный pydantic-core на Rust. Это смена архитектуры: быстрее, строже и с новой API-парадигмой для валидации и сериализации.

🔧 Главное изменение — API и ядро

🟢 Вся валидация теперь идёт через pydantic-core (Rust) — профит по скорости и надёжности.
🟢 Старые удобные алиасы и методы переехали/переименованы: parse_obj → model_validate, dict() → model_dump/model_dump_json.


📦 Простая модель и валидация (новый базовый паттерн)

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str = "Anon"

u = User.model_validate({"id": "1", "name": "Bob"})
print(u)  # id=1  (если режим lax — pydantic попытается привести)

➡️ model_validate — основной метод для валидации словарей/объектов.

⚡️ Строгость vs приведение типов

По умолчанию Pydantic пытается привести типы. Но ты можешь включить строгий режим на модель:

from pydantic import BaseModel, ConfigDict

class M(BaseModel):
    model_config = ConfigDict(strict=True)
    id: int

M.model_validate({"id": "1"})  # ValidationError: строка → int запрещена

➡️ model_config = ConfigDict(strict=True) включает strict-режим для всех полей модели.

Если нужен только жёсткий тип для конкретного поля — используй строгие типы (StrictInt, StrictStr и т.д.).

🧪 Поля и кастомная валидация — @field_validator

from pydantic import BaseModel, field_validator, ValidationError

class User(BaseModel):
    username: str
    password: str
    password2: str

    @field_validator('password2')
    def passwords_match(cls, v, info):
        if v != info.data['password']:
            raise ValueError("Пароли не совпадают")
        return v

try:
    User.model_validate({'username': 'u', 'password': 'x', 'password2': 'y'})
except ValidationError as e:
    print(e)

➡️ @field_validator — заменитель старого @validator. Может запускаться before/after и получать доступ к уже валидированным данным через ValidationInfo.


🔗 Валидация уровня модели — @model_validator

from pydantic import BaseModel, model_validator

class Payment(BaseModel):
    amount: int
    currency: str

    @model_validator(mode='after')
    def check_amount(cls, values):
        if values['amount'] <= 0:
            raise ValueError("amount must be > 0")
        return values

➡️ @model_validator — для cross-field правил и общего pre/post-processing. Есть режимы before, after, wrap.

🔁 Создание без валидации (performance / trust)

user = User.model_construct({"id": 1, "name": "x"})

➡️ model_construct() создаёт экземпляр без валидации — только если данные доверенные и нужны микро-оптимизации.

🧾 Сериализация: model_dump / model_dump_json

u = User.model_validate({"id": 1, "name": "Bob"})
print(u.model_dump())           # dict
print(u.model_dump_json())      # json
# опции: exclude_none, exclude_unset, round_trip, by_alias и т.д.

➡️ model_dump заменил dict() и даёт контроль над сериализацией.

🚫 Extra поля и конфигурация поведения

class A(BaseModel):
    model_config = ConfigDict(extra='forbid')
    name: str

A.model_validate({'name': 'x', 'bad': 1})  # ValidationError, extra not allowed

➡️ model_config.extra — ignore (по умолчанию), allow, forbid.

❌/✔️ Частые миграционные ловушки

❌ Привычка: User.parse_obj(...)
✔️ Теперь: User.model_validate(...).

❌ Ожидать, что dict() сохранит все флаги и алиасы.
✔️ Используй model_dump(by_alias=True, exclude_unset=True).

❌ Делать сложную cross-field валидацию в @field_validator.
✔️ Для таких случаев — @model_validator(mode='after').

🛠 Практические паттерны и советы

👍 Нужна производительность? Для доверенных данных используй model_construct().
👍 Хочешь полностью выключить «магическое» приведение — model_config = ConfigDict(strict=True) или StrictInt/StrictStr.
👍 Для сложных преобразований — @field_validator(..., mode='before'), чтобы нормализовать вход до основной валидации.

🗣️ Запомни: Pydantic v2 — это тот же Pydantic, но с новой, более строгой и быстрой сердцевиной.

👩‍💻 Чтение и запись файлов в Python: Легко и просто за 5 минут!

В этом видео автор простым и наглядным языком показывает, как открывать, читать и записывать файлы в Python: режимы `r/w/a`, контекстный менеджер `with`, методы `read()`, `readlines()` и `write()`, а также базовые приёмы обработки ошибок при работе с файлами.

🗣️Запомни: использование правильных режимов и `with` предотвращает утечки ресурсов и делает работу с файлами безопасной.

➡️ Ссылка на первоисточник

🤩 Pytstart || #Видеокурс

🚰 Стриминг больших данных через генераторы и yield from

Когда данных слишком много — миллионы строк в файле, бесконечные события или поток логов — держать всё в памяти нельзя. Тут спасают генераторы и ключевое слово yield from, позволяющее строить цепочки ленивых итераторов.

📦 Чтение файла построчно

def read_file(path):
    with open(path, "r") as f:
        for line in f:
            yield line.strip()

for row in read_file("bigdata.txt"):
    print(row)

➡️ Память не переполняется: читается и отдаётся по строке.

🔗 Комбинация генераторов через yield from

def numbers():
    yield from range(5)
    yield from range(10, 15)

print(list(numbers()))

➡️ yield from разворачивает подгенераторы и прокидывает их значения наружу.

⚡️ Обработка пайплайнами

def read_lines(path):
    with open(path) as f:
        yield from f

def filter_errors(lines):
    for line in lines:
        if "ERROR" in line:
            yield line

errors = filter_errors(read_lines("app.log"))
for e in errors:
    print(e)

➡️ Логика разбивается на звенья, и каждое обрабатывает поток данных лениво.

🌀 Бесконечные источники

import time

def ticker():
    while True:
        yield time.time()
        time.sleep(1)

for t in ticker():
    print(t)

➡️ Генератор выдаёт значения бесконечно, но потребитель сам решает, когда остановиться.

📑 Вложенные пайплайны с yield from

def split_words(lines):
    for line in lines:
        yield from line.split()

data = ["hello world", "big data stream"]
for word in split_words(data):
    print(word)

➡️ yield from красиво разрывает вложенные итерации без двойных циклов.

🚀 Генераторы + итераторы стандартной библиотеки

from itertools import islice

def stream():
    for i in range(1000000):
        yield i

print(list(islice(stream(), 5)))

➡️ Вместо списка в миллион чисел получаешь поток, из которого берёшь только нужное.

🗣️ Запомни: генераторы — это ленивые конвейеры данных, yield from связывает их в цепочку. Такой подход позволяет обрабатывать гигабайты без переполнения памяти и писать код, который дышит как поток.

🚦 Python + Redis Streams — быстрая обработка событий

Redis Streams — это структура данных для очередей и событий. С ней можно строить системы, где сообщения не теряются, а обрабатываются быстро и упорядоченно.

📥 Запись события в стрим

import redis
r = redis.Redis()

r.xadd("mystream", {"user": "ivan", "action": "login"})

➡️ Каждое событие получает уникальный ID и хранится в очереди.

📤 Чтение из стрима

events = r.xread({"mystream": "0"}, count=5)
print(events)

➡️ Можно читать все сообщения с начала или только новые.

👥 Консьюмер-группы

r.xgroup_create("mystream", "group1", id="0", mkstream=True)
msg = r.xreadgroup("group1", "consumer-1", {"mystream": ">"}, count=1)

➡️ Несколько воркеров обрабатывают события параллельно, но каждое сообщение достанется только одному.

♻️ Подтверждение обработки

r.xack("mystream", "group1", msg_id)

➡️ Сообщение убирается из «висячих», значит оно точно обработано.

🔄 Повторная доставка

Если воркер упал, Redis отдаст «неподтверждённые» события другому.

➡️ Это делает систему надёжной.

⚡️ Потоковая обработка в реальном времени

while True:
    msgs = r.xread({"mystream": "$"}, block=5000)
    for _, events in msgs:
        for _, data in events:
            print(data)

➡️ Код крутится бесконечно и реагирует на новые события сразу.

🚀 Кейсы применения

Логи, чаты, системы мониторинга, стриминг данных с IoT-устройств. Всё, где важна скорость и порядок.

🗣️ Запомни: Redis Streams позволяют строить быстрые и надёжные пайплайны обработки событий, а с consumer groups ты получаешь масштабируемость и защиту от потерь.

📐 Typing: TypedDict, Literal, Annotated в реальных проектах

Аннотации типов в Python — это не только int и str. Для сложных сценариев есть мощные инструменты: TypedDict, Literal, Annotated. Они помогают писать код понятнее и ловить ошибки ещё на этапе разработки.

📦 TypedDict — словари со строгой схемой

from typing import TypedDict

class User(TypedDict):
    id: int
    name: str
    is_admin: bool

u: User = {"id": 1, "name": "Иван", "is_admin": False}

➡️ Теперь IDE и mypy знают структуру словаря. Ошибку в ключе или типе отловишь сразу.

🎯 Literal — фиксированные значения

from typing import Literal

def set_status(status: Literal["NEW", "IN_PROGRESS", "DONE"]) -> None:
    print(f"Статус: {status}")

set_status("NEW")          # ✅
set_status("INVALID")      # ❌ mypy отловит

➡️ Полезно, когда параметр должен принимать строго ограниченные значения.

🧩 Annotated — тип + метаданные

from typing import Annotated
from dataclasses import dataclass

@dataclass
class User:
    id: Annotated[int, "DB Primary Key"]
    email: Annotated[str, "must be valid email"]

➡️ Аннотации становятся документацией и могут использоваться библиотеками (например, валидацией в FastAPI).

⚡️ TypedDict с опциональными ключами

class Config(TypedDict, total=False):
    debug: bool
    log_level: str

cfg: Config = {"debug": True}  # ok

➡️ total=False позволяет делать часть ключей необязательными.

🔄 Literal в связке с Enum

from typing import Literal

Role = Literal["user", "admin", "moderator"]

def check_role(role: Role): ...

➡️ Даёт компактную альтернативу Enum, если не нужно хранить методы.

📑 Annotated для валидации данных
В FastAPI:

from typing import Annotated
from fastapi import Query

def read_items(size: Annotated[int, Query(ge=1, le=100)]):
    return {"size": size}

➡️Ограничения накладываются прямо через типы, без лишнего кода.

🧠 Совмещение инструментов

class Product(TypedDict):
    id: int
    status: Literal["active", "archived"]

def process(p: Product): ...

➡️ Получается словарь со строгой схемой и фиксированным набором значений.

🗣️ Запомни: TypedDict даёт строгие словари, Literal ограничивает набор значений, Annotated связывает тип с метаданными. Вместе они делают код самодокументируемым и надёжным.

👩‍💻 Командная строка в Python | Модуль os

В этом видео автор объясняет, как работать с модулем os в Python для взаимодействия с операционной системой.
Разбирается выполнение команд через терминал, навигация по файловой системе, работа с путями и управление файлами. Всё показано на простых примерах, которые легко повторить самостоятельно.

🗣️Запомни: модуль os открывает доступ к системным возможностям прямо из Python, делая код универсальнее и мощнее.

➡️ Ссылка на первоисточник

🤩 Pytstart || #Видеокурс

Async ORM в Python: SQLModel/Tortoise 🆚 SQLAlchemy

Синхронные ORM вроде SQLAlchemy блокируют потоки на время запросов к БД — в асинхронных приложениях это убивает всю пользу asyncio. Async ORM работают на корутинах и не блокируют event loop, позволяя обрабатывать тысячи одновременных подключений.

🔄 SQLAlchemy: синхронный подход (блокирующий)

# ❌ Блокирует поток на время выполнения запроса
from sqlalchemy.orm import Session

def get_users_sync():
    with Session(engine) as session:
        users = session.query(User).all()  # Поток заблокирован
        return users

➡️ Каждый запрос занимает поток. В async-приложениях нужны тред-пулы, что усложняет архитектуру.

⚡️ SQLModel: асинхронная обёртка над SQLAlchemy

# ✅ Не блокирует event loop
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession

async def get_users_async():
    async with AsyncSession(engine) as session:
        result = await session.exec(select(User))  # Освобождает loop
        return result.all()

➡️ Полная совместимость с SQLAlchemy + асинхронность. Идеально для миграции.

🐢 Tortoise ORM: нативный async-подход

# ✅ Специально разработан для асинхронности
from tortoise.models import Model
from tortoise import fields

class User(Model):
    name = fields.CharField(max_length=255)
    email = fields.CharField(max_length=255)
    
async def get_users_native():
    return await User.all()  # Чистый async API

➡️ Вдохновлен Django ORM — простой и понятный синтаксис.

📊 Сравнение производительности

# Тест: 1000 одновременных запросов
# SQLAlchemy + threads: ~15 сек, 1.2 GB RAM
# SQLModel: ~2.8 сек, 620 MB RAM
# Tortoise: ~2.3 сек, 512 MB RAM

➡️ Async ORM в 5-7 раз быстрее при высокой нагрузке благодаря отсутствию блокировок.

🛠 Работа с отношениями: SQLModel

class Team(SQLModel, table=True):
    id: int | None = Field(primary_key=True)
    name: str
    users: List["User"] = Relationship(back_populates="team")

class User(SQLModel, table=True):
    team_id: int | None = Field(foreign_key="team.id")
    team: Optional[Team] = Relationship(back_populates="users")

➡️ Полная поддержка отношений как в SQLAlchemy + Pydantic валидация.

🛠 Работа с отношениями: Tortoise

class Team(Model):
    name = fields.CharField(max_length=255)
    users: fields.ReverseRelation["User"]

class User(Model):
    name = fields.CharField(max_length=255)
    team: fields.ForeignKeyRelation[Team] = fields.ForeignKeyField(
        "models.Team", related_name="users"
    )

➡️ Простой Django-like синтаксис с автоматическими обратными связями.

⚡️ Миграции и инструменты
🟢 SQLAlchemy: Требует Alembic, сложная настройка
🟢 SQLModel: Работает через Alembic, наследует сложности SQLAlchemy
🟢 Tortoise: Встроенная система миграций, простая настройка

🔧 Экосистема и сообщество

🟢 SQLAlchemy: Огромное сообщество, все возможные расширения
🟢 SQLModel: Молодой, но растущий проект, полная совместимость
🟢 Tortoise: Зрелый async-ориентированный фреймворк

🌐 Использование с FastAPI

# SQLModel интеграция
from fastapi import FastAPI
from sqlmodel.ext.asyncio.session import AsyncSession

app = FastAPI()

@app.get("/users")
async def get_users(session: AsyncSession = Depends(get_session)):
    result = await session.exec(select(User))
    return result.all()

➡️ Идеальная интеграция с современными async-фреймворками.

🗣️ Запомни: Выбирай Tortoise для новых async-проектов за его простоту и производительность.

🖥 import - механизм подключения модулей

В Python import используется для подключения модулей (файлов с кодом) в текущую программу.

1. Импорт всего модуля

import math
print(math.sqrt(16))  # 4.0

2. Импорт с псевдонимом (alias)

import numpy as np
print(np.pi)  # 3.141592653589793

3. Импорт конкретных объектов

from math import sqrt, pi
print(sqrt(25))  # 5.0
print(pi)  # 3.141592653589793

4. Импорт всех объектов (*) ❗ (не рекомендуется)

from math import *
print(sin(0))  # 0.0

⚠️ Минус: может вызвать конфликты имён.

5. Импорт из пользовательского модуля (my_module.py)

import my_module  
my_module.hello()

Если модуль в другой папке, нужно добавить его в sys.path:

import sys
sys.path.append("path/to/module")
import my_module

Python ищет модуль в следующем порядке:
1️⃣ Стандартные модули (math, random, os)
2️⃣ Текущая директория (папка, где запускается скрипт)
3️⃣ Пути из sys.path

Если модуль не найден, возникает ModuleNotFoundError.

🗣️ import позволяет использовать готовый код и организовать программу, разбивая её на модули.