🔠 Type Hints: код говорит за тебя
Python — динамический, можешь писать без типов.
Но с аннотациями
Типы не проверяются в рантайме — но помогают читать, дебажить и ловить ошибки раньше.
📛 Пример: аннотации в функции
➡️ В рантайме работает так же.
Но теперь IDE и mypy понимают: на входе и выходе только числа.
🧪 Комбинации типов
➡️ Python по-прежнему не строгий.
Но твой код стал декларативным: видно, что ждёт функция.
📦 Опциональные значения
➡️ Без
А с
🔐 Собственные типы
➡️
но для анализатора — это разные сущности.
🪄 Магия PEP 604 (Python 3.10+)
➡️ Больше не нужно
Теперь типы можно писать через
🗣️ Запомни: 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}
Но твой код стал декларативным: видно, что ждёт функция.
📦 Опциональные значения
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 одинаковы в рантайме,но для анализатора — это разные сущности.
def foo(x: int | str) -> None:
...
Union.Теперь типы можно писать через
|.Please open Telegram to view this post
VIEW IN TELEGRAM
❤6👍3🔥3⚡1👎1🍌1
This media is not supported in your browser
VIEW IN TELEGRAM
while, Операторы break и continueВ этом видео автор подробно объясняет, как работает цикл while в Python.
Разбираются операторы break и continue, показано, как с их помощью управлять выполнением цикла. Всё демонстрируется на простых примерах, что помогает легко понять и применить эти конструкции на практике.
Please open Telegram to view this post
VIEW IN TELEGRAM
👍7🔥6❤1
🧵
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')Please open Telegram to view this post
VIEW IN TELEGRAM
🔥7❤3⚡2👍1
Ты написал:
def greet(name):
print(f"Привет, {name}!")
А потом захотел добавить логирование, проверку ошибок или права доступа. Переписывать код не хочется.
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
Please open Telegram to view this post
VIEW IN TELEGRAM
❤6🔥3⚡1
__str____repr__ : зачем два разных отображения объектаВ Python у любого объекта можно определить два разных способа отображения —
__str__ и __repr__. Кажется избыточным, но на деле это основа читаемости и отладки.Разберём всё по полочкам
class User:
def __init__(self, name):
self.name = name
def __str__(self):
return f"Пользователь {self.name}"
u = User("Иван")
print(u) # Пользователь Иван
__str__ нужен для красивого вывода в консоли, логах или интерфейсе. Его цель — понятность.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__ всегда должен быть однозначным. Его читают программисты и дебаггер, а не пользователи.repru = User("Иван")
u__repr__. Поэтому именно он важен для отладки.📑 Логи: str для людей, repr для разработчиков
print(str(u)) # Пользователь Иван
print(repr(u)) # User(name='Иван')
class Item:
def __repr__(self):
return "Item(id=42)"
i = Item()
print(i) # Item(id=42)
__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.Грамотное разделение делает код удобнее и чище, а логи — понятнее.Please open Telegram to view this post
VIEW IN TELEGRAM
👍7❤3🔥2⚡1👏1
UnboundLocalError ломает кодВ Python переменные живут в разных областях видимости: локальной (функция), глобальной (модуль), и встроенной (
len, print, и т. д.).Новички часто сталкиваются с
UnboundLocalError — когда кажется, что переменная доступна, но интерпретатор думает иначе.
x = 10
def foo():
print(x) # 🔴 UnboundLocalError
x = 5
foo()
x = 5 и считает, что x — локальная переменная. Но до присвоения её нет, поэтому ошибка.
x = 10
def foo():
global x
print(x) # ✅ 10
x = 5
foo()
print(x) # 5
x.nonlocal
def outer():
x = 10
def inner():
x = x + 1 # 🔴 UnboundLocalError
return x
return inner()
outer()
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 — для замыканий. Избегай глобалок: они делают код запутанным. Лучше возвращать значения и передавать их явно.Please open Telegram to view this post
VIEW IN TELEGRAM
❤3👍3⚡2
В 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
def add_one(x=0):
return x + 1
print(add_one()) # 1
print(add_one()) # 1
arg=None, потом if arg is None: arg = []. Исключение — если тебе нужен общий объект (например, кеш).Please open Telegram to view this post
VIEW IN TELEGRAM
👍5❤4⚡2
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
🟢 обработка логов и CSV гигабайтного размера;🟢 потоковые данные (сокеты, API);🟢 построение бесконечных последовательностей;🟢 конвейеры обработки (map/filter без лишних списков).
yield отдаёт результат по частям, не занимая память. Генераторы можно объединять в цепочки — как трубы. Если данных много или они бесконечны — только генераторы спасут.Please open Telegram to view this post
VIEW IN TELEGRAM
🔥7🤝3❤2
This media is not supported in your browser
VIEW IN TELEGRAM
В этом видео автор объясняет, как создавать функции в Python с помощью ключевого слова def и использовать оператор return для возврата результата. Показано, зачем нужны функции, как они упрощают код и делают его удобным для повторного использования.
Please open Telegram to view this post
VIEW IN TELEGRAM
❤5⚡3
This media is not supported in your browser
VIEW IN TELEGRAM
В этом видео разбирается, что такое модули в Python и как их подключать в проект. Автор показывает, как импортировать встроенные модули, а также создавать и использовать свои собственные. Это позволяет делать код более структурированным и переиспользуемым.
Please open Telegram to view this post
VIEW IN TELEGRAM
❤🔥2❤2⚡2
🚦 Python + Redis Streams — быстрая обработка событий
Redis Streams — это структура данных для очередей и событий. С ней можно строить системы, где сообщения не теряются, а обрабатываются быстро и упорядоченно.
📥 Запись события в стрим
➡️ Каждое событие получает уникальный ID и хранится в очереди.
📤 Чтение из стрима
➡️ Можно читать все сообщения с начала или только новые.
👥 Консьюмер-группы
➡️ Несколько воркеров обрабатывают события параллельно, но каждое сообщение достанется только одному.
♻️ Подтверждение обработки
➡️ Сообщение убирается из «висячих», значит оно точно обработано.
🔄 Повторная доставка
➡️ Это делает систему надёжной.
⚡️ Потоковая обработка в реальном времени
➡️ Код крутится бесконечно и реагирует на новые события сразу.
🚀 Кейсы применения
🗣️ Запомни: Redis Streams позволяют строить быстрые и надёжные пайплайны обработки событий, а с consumer groups ты получаешь масштабируемость и защиту от потерь.
Redis Streams — это структура данных для очередей и событий. С ней можно строить системы, где сообщения не теряются, а обрабатываются быстро и упорядоченно.
import redis
r = redis.Redis()
r.xadd("mystream", {"user": "ivan", "action": "login"})
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-устройств. Всё, где важна скорость и порядок.
Please open Telegram to view this post
VIEW IN TELEGRAM
👍4❤2🥰2
Pydantic v2 — не мелкий апдейт, а рефакторинг движка в отдельный
pydantic-core на Rust. Это смена архитектуры: быстрее, строже и с новой 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 — основной метод для валидации словарей/объектов.По умолчанию 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_validatorfrom 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_validatorfrom 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.user = User.model_construct({"id": 1, "name": "x"})model_construct() создаёт экземпляр без валидации — только если данные доверенные и нужны микро-оптимизации.model_dump / model_dump_jsonu = 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() и даёт контроль над сериализацией.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'), чтобы нормализовать вход до основной валидации.
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥7❤2👍2❤🔥1
Media is too big
VIEW IN TELEGRAM
В этом видео автор простым и наглядным языком показывает, как открывать, читать и записывать файлы в Python: режимы `r/w/a`, контекстный менеджер `with`, методы `read()`, `readlines()` и `write()`, а также базовые приёмы обработки ошибок при работе с файлами.
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥4❤2
🚰 Стриминг больших данных через генераторы и
Когда данных слишком много — миллионы строк в файле, бесконечные события или поток логов — держать всё в памяти нельзя. Тут спасают генераторы и ключевое слово
📦 Чтение файла построчно
➡️ Память не переполняется: читается и отдаётся по строке.
🔗 Комбинация генераторов через yield from
➡️
⚡️ Обработка пайплайнами
➡️ Логика разбивается на звенья, и каждое обрабатывает поток данных лениво.
🌀 Бесконечные источники
➡️ Генератор выдаёт значения бесконечно, но потребитель сам решает, когда остановиться.
📑 Вложенные пайплайны с
➡️
🚀 Генераторы + итераторы стандартной библиотеки
➡️ Вместо списка в миллион чисел получаешь поток, из которого берёшь только нужное.
🗣️ Запомни: генераторы — это ленивые конвейеры данных, yield from связывает их в цепочку. Такой подход позволяет обрабатывать гигабайты без переполнения памяти и писать код, который дышит как поток.
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)
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 fromdef 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)))
Please open Telegram to view this post
VIEW IN TELEGRAM
❤6⚡1
🚦 Python + Redis Streams — быстрая обработка событий
Redis Streams — это структура данных для очередей и событий. С ней можно строить системы, где сообщения не теряются, а обрабатываются быстро и упорядоченно.
📥 Запись события в стрим
➡️ Каждое событие получает уникальный ID и хранится в очереди.
📤 Чтение из стрима
➡️ Можно читать все сообщения с начала или только новые.
👥 Консьюмер-группы
➡️ Несколько воркеров обрабатывают события параллельно, но каждое сообщение достанется только одному.
♻️ Подтверждение обработки
➡️ Сообщение убирается из «висячих», значит оно точно обработано.
🔄 Повторная доставка
➡️ Это делает систему надёжной.
⚡️ Потоковая обработка в реальном времени
➡️ Код крутится бесконечно и реагирует на новые события сразу.
🚀 Кейсы применения
🗣️ Запомни: Redis Streams позволяют строить быстрые и надёжные пайплайны обработки событий, а с consumer groups ты получаешь масштабируемость и защиту от потерь.
Redis Streams — это структура данных для очередей и событий. С ней можно строить системы, где сообщения не теряются, а обрабатываются быстро и упорядоченно.
import redis
r = redis.Redis()
r.xadd("mystream", {"user": "ivan", "action": "login"})
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-устройств. Всё, где важна скорость и порядок.
Please open Telegram to view this post
VIEW IN TELEGRAM
❤6👍4
📐 Typing: TypedDict, Literal, Annotated в реальных проектах
Аннотации типов в Python — это не только
📦 TypedDict — словари со строгой схемой
➡️ Теперь IDE и mypy знают структуру словаря. Ошибку в ключе или типе отловишь сразу.
🎯 Literal — фиксированные значения
➡️ Полезно, когда параметр должен принимать строго ограниченные значения.
🧩 Annotated — тип + метаданные
➡️ Аннотации становятся документацией и могут использоваться библиотеками (например, валидацией в FastAPI).
⚡️ TypedDict с опциональными ключами
➡️
🔄 Literal в связке с Enum
➡️ Даёт компактную альтернативу Enum, если не нужно хранить методы.
📑 Annotated для валидации данных
В FastAPI:
➡️ Ограничения накладываются прямо через типы, без лишнего кода.
🧠 Совмещение инструментов
➡️ Получается словарь со строгой схемой и фиксированным набором значений.
🗣️ Запомни: TypedDict даёт строгие словари, Literal ограничивает набор значений, Annotated связывает тип с метаданными. Вместе они делают код самодокументируемым и надёжным.
Аннотации типов в Python — это не только
int и str. Для сложных сценариев есть мощные инструменты: TypedDict, Literal, Annotated. Они помогают писать код понятнее и ловить ошибки ещё на этапе разработки.from typing import TypedDict
class User(TypedDict):
id: int
name: str
is_admin: bool
u: User = {"id": 1, "name": "Иван", "is_admin": False}
from typing import Literal
def set_status(status: Literal["NEW", "IN_PROGRESS", "DONE"]) -> None:
print(f"Статус: {status}")
set_status("NEW") # ✅
set_status("INVALID") # ❌ mypy отловит
from typing import Annotated
from dataclasses import dataclass
@dataclass
class User:
id: Annotated[int, "DB Primary Key"]
email: Annotated[str, "must be valid email"]
⚡️ TypedDict с опциональными ключами
class Config(TypedDict, total=False):
debug: bool
log_level: str
cfg: Config = {"debug": True} # ok
total=False позволяет делать часть ключей необязательными.from typing import Literal
Role = Literal["user", "admin", "moderator"]
def check_role(role: Role): ...
📑 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): ...
Please open Telegram to view this post
VIEW IN TELEGRAM
👍4❤2🔥2
Media is too big
VIEW IN TELEGRAM
В этом видео автор объясняет, как работать с модулем os в Python для взаимодействия с операционной системой.
Разбирается выполнение команд через терминал, навигация по файловой системе, работа с путями и управление файлами. Всё показано на простых примерах, которые легко повторить самостоятельно.
Please open Telegram to view this post
VIEW IN TELEGRAM
❤3⚡2
Async ORM в Python: SQLModel/Tortoise 🆚 SQLAlchemy
Синхронные ORM вроде SQLAlchemy блокируют потоки на время запросов к БД — в асинхронных приложениях это убивает всю пользу asyncio. Async ORM работают на корутинах и не блокируют event loop, позволяя обрабатывать тысячи одновременных подключений.
🔄 SQLAlchemy: синхронный подход (блокирующий)
➡️ Каждый запрос занимает поток. В async-приложениях нужны тред-пулы, что усложняет архитектуру.
⚡️ SQLModel: асинхронная обёртка над SQLAlchemy
➡️ Полная совместимость с SQLAlchemy + асинхронность. Идеально для миграции.
🐢 Tortoise ORM: нативный async-подход
➡️ Вдохновлен Django ORM — простой и понятный синтаксис.
📊 Сравнение производительности
➡️ Async ORM в 5-7 раз быстрее при высокой нагрузке благодаря отсутствию блокировок.
🛠 Работа с отношениями: SQLModel
➡️ Полная поддержка отношений как в SQLAlchemy + Pydantic валидация.
🛠 Работа с отношениями: Tortoise
➡️ Простой Django-like синтаксис с автоматическими обратными связями.
🔧 Экосистема и сообщество
🌐 Использование с FastAPI
➡️ Идеальная интеграция с современными async-фреймворками.
🗣️ Запомни: Выбирай Tortoise для новых async-проектов за его простоту и производительность.
Синхронные ORM вроде SQLAlchemy блокируют потоки на время запросов к БД — в асинхронных приложениях это убивает всю пользу asyncio. Async ORM работают на корутинах и не блокируют event loop, позволяя обрабатывать тысячи одновременных подключений.
# ❌ Блокирует поток на время выполнения запроса
from sqlalchemy.orm import Session
def get_users_sync():
with Session(engine) as session:
users = session.query(User).all() # Поток заблокирован
return users
# ✅ Не блокирует 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()
# ✅ Специально разработан для асинхронности
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
# Тест: 1000 одновременных запросов
# SQLAlchemy + threads: ~15 сек, 1.2 GB RAM
# SQLModel: ~2.8 сек, 620 MB RAM
# Tortoise: ~2.3 сек, 512 MB RAM
🛠 Работа с отношениями: 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")
🛠 Работа с отношениями: 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"
)
⚡️ Миграции и инструменты🟢 SQLAlchemy: Требует Alembic, сложная настройка🟢 SQLModel: Работает через Alembic, наследует сложности SQLAlchemy🟢 Tortoise: Встроенная система миграций, простая настройка
🟢 SQLAlchemy: Огромное сообщество, все возможные расширения🟢 SQLModel: Молодой, но растущий проект, полная совместимость🟢 Tortoise: Зрелый async-ориентированный фреймворк
# 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()
Please open Telegram to view this post
VIEW IN TELEGRAM
❤2👍2
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 позволяет использовать готовый код и организовать программу, разбивая её на модули.Please open Telegram to view this post
VIEW IN TELEGRAM
⚡1👍1
HTTP/3 работает поверх QUIC (UDP). Меньше задержек, параллельные стримы, нет head-of-line блокировки. Для Python это новая эра веб-сервисов.
from aioquic.asyncio.client import connect
async with connect("example.com", 443, alpn_protocols=["h3"]) as client:
stream_id = client.get_next_available_stream_id()
client.send_headers(stream_id, [(":method", "GET"), (":path", "/")])
client.send_eof(stream_id)
📡 Сервер на aioquic
from aioquic.asyncio import serve
from aioquic.h3.connection import H3_ALPN
async def handler(stream_id, request):
return [(b":status", b"200")], b"Hello QUIC!"
await serve("0.0.0.0", 4433, configuration=my_tls_config,
alpn_protocols=H3_ALPN, stream_handler=handler)
QUIC = несколько потоков внутри одного соединения.
id1 = client.get_next_available_stream_id()
id2 = client.get_next_available_stream_id()
client.send_headers(id1, [...])
client.send_headers(id2, [...])
QUIC всегда шифрован.
Пока через uvicorn-h3 (экспериментально).
uvicorn app:app --http h3 --port 4433
🟢 API с большим количеством параллельных запросов🟢 real-time (чаты, стримы)🟢 мобильные клиенты с нестабильной сетью
❌ Минусы
🔴 Поддержка в браузерах уже есть, но в Python всё ещё эксперимент.🔴 aioquic — низкоуровневый, нужно писать руками.🔴 Поддержка в фреймворках только зарождается.
Please open Telegram to view this post
VIEW IN TELEGRAM
👍5🔥2❤1
Графы — не про математику, а про связи. Пользователи, друзья, транзакции, сервисы — всё это графы. Python и современные БД делают их работу простой и мощной.
Создадим и изучим связи между объектами:
import networkx as nx
G = nx.Graph()
G.add_edges_from([('Alice', 'Bob'), ('Bob', 'Eve'), ('Alice', 'Eve')])
print(nx.shortest_path(G, 'Alice', 'Eve'))
print(nx.degree_centrality(G))
import matplotlib.pyplot as plt
nx.draw(G, with_labels=True, node_color='lightblue', node_size=1500)
plt.show()
GraphQL от Facebook — не база, а язык запросов к API.
query = """
{
user(id: "1") {
name
friends {
name
}
}
}
"""
from ariadne import QueryType, make_executable_schema, graphql_sync
type_defs = """
type Query {
hello: String!
}
"""
query = QueryType()
@query.field("hello")
def resolve_hello(_, info):
return "GraphQL + Python!"
schema = make_executable_schema(type_defs, query)
Python-драйвер для Neo4j:
from neo4j import GraphDatabase
driver = GraphDatabase.driver("bolt://localhost:7687")
with driver.session() as session:
result = session.run("MATCH (a)-[r]->(b) RETURN a,b LIMIT 5")
for record in result:
print(record)
MATCH (user:Person)-[:FRIEND]->(friend)
WHERE user.name = "Alice"
RETURN friend.name
Анализируешь связи в NetworkX, сохраняешь их в Neo4j, а отдаёшь клиенту через GraphQL API.
Полный цикл: анализ → хранение → доступ.
Please open Telegram to view this post
VIEW IN TELEGRAM
👍3❤2⚡2