PytStart | Программирование на Python
4.45K subscribers
20 photos
26 videos
54 links
Python: примеры кода, уроки, статьи

Купить рекламу: https://telega.in/c/pytstart

✍️По всем вопросам: @Pascal4eg
Download Telegram
🖥 Python-фишка: Игнорируем ошибки элегантно

Вам наверняка приходилось писать код, где ошибку нужно просто проигнорировать.
Например, удалить файл, если он есть (а если нет — ну и ладно).

Обычно это выглядит так:

import os

try:
os.remove("temp.txt")
except FileNotFoundError:
pass # Выглядит не очень красиво


Pythonic way:
Используйте контекстный менеджер suppress из модуля contextlib.


import os
from contextlib import suppress

with suppress(FileNotFoundError):
os.remove("temp.txt")


Это делает намерение программиста очевидным: "Попробуй выполнить, но если вылетит эта ошибка — подави её". Код становится чище и читаемее.

#python #cleancode #refactoring #tips
Please open Telegram to view this post
VIEW IN TELEGRAM
👍53👎2
🖥 Python-фишка: Переворачиваем словарь одной строкой

Допустим, у вас есть словарь, и нужно поменять местами ключи и значения (инвертировать его).

Исходные данные:

currencies = {'USD': 1, 'EUR': 0.9, 'GBP': 0.8}


Вместо того чтобы создавать пустой словарь и запускать цикл, используйте Dict Comprehension:


# {значение: ключ}
inverted = {value: key for key, value in currencies.items()}

print(inverted)
# {1: 'USD', 0.9: 'EUR', 0.8: 'GBP'}


Синтаксис {k: v for ...} работает так же мощно, как и для списков. Можно даже добавить условия:

# Оставим только валюты с курсом меньше 1
cheap = {k: v for k, v in currencies.items() if v < 1}
Please open Telegram to view this post
VIEW IN TELEGRAM
4👍3
🖥 Ускорение кода в 100 раз одной строкой

Если у вас есть функция, которая выполняет тяжелые вычисления (или рекурсию) с одними и теми же аргументами, вы можете мгновенно её ускорить.

Импортируем декоратор `@lru_cache` из модуля functools.

Пример (классический Фибоначчи):


from functools import lru_cache

@lru_cache(maxsize=None) # <--- Вся магия здесь
def fib(n):
if n < 2:
return n
return fib(n-1) + fib(n-2)

print(fib(50))


Без кэша: Python будет считать это вечность (миллиарды вызовов).
С кэшем: Результат выведется мгновенно.

Декоратор запоминает результаты вызова функции. Если функцию вызовут снова с теми же аргументами, Python не будет считать заново, а просто возьмет готовый ответ из памяти.

#python #performance #optimization #functools
Please open Telegram to view this post
VIEW IN TELEGRAM
6
🖥 Именованные кортежи

Обычные кортежи (`tuple`) экономят память, но обращаться к данным по индексу — неудобно.
point[0] — что это? Координата X? Широта? ID пользователя?

Используйте `NamedTuple` из модуля typing (или `collections`). Это как класс, только легче.


from typing import NamedTuple

class Point(NamedTuple):
x: int
y: int

p = Point(10, 20)

# Теперь можно обращаться по имени!
print(p.x, p.y) # 10 20

# Но поведение кортежа сохраняется
print(p[0]) # 10


Это делает код "самодокументируемым". Идеально для возврата нескольких значений из функции.

#python #typing #structures #tips
Please open Telegram to view this post
VIEW IN TELEGRAM
6
🤯 Математика с булевыми значениями

А вы знали, что в Python True и False — это на самом деле просто замаскированные 1 и 0? Тип bool наследуется от int.

Поэтому такой код абсолютно валиден:


x = True + True + 5
print(x)
# Вывод: 7 (1 + 1 + 5)


Или даже так (используем как индекс):

options = ["Нет", "Да"]
is_agreed = True

print(options[is_agreed])
# Вывод: "Да" (потому что options[1])


💡 Примечание: в продакшене лучше так не писать, но знать об этом полезно!

#python #funfacts #underTheHood
Please open Telegram to view this post
VIEW IN TELEGRAM
👍83
Media is too big
VIEW IN TELEGRAM
👩‍💻 Работа с внешними API на Python: библиотека requests

В этом видео автор на практике разбирает, как взаимодействовать с внешними сервисами и данными через API. Вы научитесь отправлять GET и POST-запросы, обрабатывать ответы в формате JSON, а также работать с заголовками и параметрами с помощью универсальной библиотеки requests.


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

🗣️Запомни: умение работать с API - это ключ к интеграции ваших программ с миром внешних данных и сервисов, от погоды до нейросетей.

🤩 Pytstart || #Видеокурс
Please open Telegram to view this post
VIEW IN TELEGRAM
4👍2
🗜 Сжатие данных на лету: gzip, lzma (работа со сжатыми файлами)

Когда логи занимают гигабайты, а дампы БД не помещаются на диск, сжатие становится must-have. Но не нужно сжимать вручную — Python умеет работать со сжатыми файлами напрямую, как с обычными.

⚡️ gzip: стандарт для логов и JSON

Модуль gzip даёт тот же интерфейс, что и open(), но прозрачно сжимает:
import gzip
import json

# ПИШЕМ сжатый JSON одной строкой
with gzip.open('data.json.gz', 'wt', encoding='utf-8') as f:
json.dump({'users': [{'id': i, 'name': f'user_{i}'} for i in range(10000)]}, f)
# Файл data.json.gz весит в 5-10 раз меньше

# ЧИТАЕМ как обычный файл
with gzip.open('data.json.gz', 'rt', encoding='utf-8') as f:
data = json.load(f)
print(f"Загружено {len(data['users'])} пользователей") # 10000

➡️ Режимы: 'rt'/'wt' для текста, 'rb'/'wb' для бинарных данных. Сжатие происходит на лету — не нужно держать в памяти весь несжатый файл.

lzma: максимальное сжатие (но медленнее)

Для архивов, где важна степень сжатия, а не скорость:
import lzma
import pickle

# Сохраняем сжатый pickle
data = [{'x': i, 'y': i**2} for i in range(100000)]

with lzma.open('data.pickle.xz', 'wb') as f:
pickle.dump(data, f)
print("Файл сохранён с максимальным сжатием")

# Загружаем обратно
with lzma.open('data.pickle.xz', 'rb') as f:
loaded = pickle.load(f)
print(f"Загружено {len(loaded)} объектов")

➡️ lzma (формат .xz) сжимает лучше gzip, но работает медленнее. Идеально для долгосрочного хранения дампов.

🔍 Сравниваем gzip vs lzma на практике
import gzip, lzma, os
import random

# Генерируем тестовые данные (повторяющиеся строки хорошо сжимаются)
text = '\n'.join(str(random.randint(1000, 9999)) for _ in range(50000))

# Сохраняем в разных форматах
with open('raw.txt', 'w') as f:
f.write(text)

with gzip.open('test.gz', 'wt') as f:
f.write(text)

with lzma.open('test.xz', 'wt') as f:
f.write(text)

# Сравниваем размеры
for name in ['raw.txt', 'test.gz', 'test.xz']:
print(f"{name}: {os.path.getsize(name)} байт")
# raw.txt: 345678 байт
# test.gz: 45678 байт (в ~7.5 раз меньше)
# test.xz: 23456 байт (в ~14.7 раз меньше)

➡️ lzma сжимает в 2-3 раза эффективнее gzip, но занимает больше CPU и времени. Выбирай по ситуации.

🧪 Контекстный менеджер для прозрачного сжатия

Создадим универсальный инструмент, который сам определяет формат по расширению:
import bz2  # ещё один формат сжатия

def smart_open(filename, mode='r', **kwargs):
"""Открывает файл, автоматически определяя сжатие по расширению."""
if filename.endswith('.gz'):
return gzip.open(filename, mode, **kwargs)
elif filename.endswith('.xz') or filename.endswith('.lzma'):
return lzma.open(filename, mode, **kwargs)
elif filename.endswith('.bz2'):
return bz2.open(filename, mode, **kwargs)
else:
return open(filename, mode, **kwargs)

# Используем единообразно
with smart_open('data.json.gz', 'rt') as f:
data = json.load(f)

with smart_open('dump.sql', 'w') as f:
f.write('SELECT * FROM users;')

➡️ Один интерфейс для всех типов файлов - код становится чище.

🚀 Потоковое сжатие сетевых данных

Сжимаем данные прямо при передаче по сети или в пайплайнах:
import gzip
import io
def compress_chunks(chunks):
"""Сжимает поток чанков на лету."""
buffer = io.BytesIO()
with gzip.GzipFile(fileobj=buffer, mode='wb') as f:
for chunk in chunks:
f.write(chunk.encode() if isinstance(chunk, str) else chunk)
buffer.seek(0)
return buffer
# Пример: сжимаем логи построчно
log_lines = [f"[INFO] Event {i}\n" for i in range(1000)]
compressed = compress_chunks(log_lines)
# Результат уже сжат
print(f"Исходно: {sum(len(l) for l in log_lines)} байт")
print(f"После сжатия: {len(compressed.getvalue())} байт")

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

🗣️ Запомни: Никогда не работай со сжатыми файлами вручную. Используй gzip.open() для быстрого сжатия логов и lzma.open() для максимального сжатия дампов. Пиши контекстные менеджеры для единого интерфейса.
Please open Telegram to view this post
VIEW IN TELEGRAM
8
Media is too big
VIEW IN TELEGRAM
👩‍💻 Обработка исключений в Python: try, except, finally

В этом видео автор подробно разбирает, как предсказуемо управлять ошибками в программах с помощью конструкции try-except. Вы научитесь перехватывать конкретные типы исключений, корректно освобождать ресурсы в блоке finally и создавать собственные классы ошибок для большей ясности кода.

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

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

🤩 Pytstart || #Видеокурс
Please open Telegram to view this post
VIEW IN TELEGRAM
👍62
🔌 Python + SQLite: скрытые возможности (возврат изменённых строк, оконные функции)

Все пишут SELECT и INSERT, но SQLite — это полноценная СУБД с фичами из больших баз. Используй RETURNING, UPSERT и оконные функции, чтобы логика работала на уровне данных, а не в коде.

⚡️ UPSERT: вставь или обнови одной командой

Классика: нужно обновить запись, если она есть, иначе создать. Раньше требовалось два запроса. Теперь — один:
import sqlite3

conn = sqlite3.connect(':memory:')
conn.execute('''CREATE TABLE users (id INTEGER PRIMARY KEY,
email TEXT UNIQUE,
login_count INTEGER)''')

# Первый вызов — вставит новую запись
conn.execute('''INSERT INTO users (email, login_count)
VALUES ('alice@mail.com', 1)
ON CONFLICT(email) DO UPDATE
SET login_count = login_count + 1''')

# Второй вызов — увеличит счётчик
conn.execute('''INSERT INTO users (email, login_count)
VALUES ('alice@mail.com', 1)
ON CONFLICT(email) DO UPDATE
SET login_count = users.login_count + 1''')

result = conn.execute('SELECT * FROM users').fetchone()
print(result) # (1, 'alice@mail.com', 2)

➡️ ON CONFLICT DO UPDATE — магия UPSERT. Ключевое поле должно иметь ограничение UNIQUE или PRIMARY KEY.

RETURNING: возвращай изменённые строки сразу

Раньше чтобы получить ID новой записи, делали отдельный SELECT. Теперь всё сразу:
# Возвращаем ID и данные новой записи
cursor = conn.execute('''INSERT INTO users (email, login_count)
VALUES ('bob@mail.com', 1)
RETURNING id, email''')
new_user = cursor.fetchone()
print(f'Новый ID: {new_user[0]}, email: {new_user[1]}')

# Работает и с UPDATE/DELETE
cursor = conn.execute('''UPDATE users
SET login_count = login_count * 2
WHERE email = 'alice@mail.com'
RETURNING email, login_count''')
updated = cursor.fetchone()
print(f'Обновлено: {updated[0]}, новые логины: {updated[1]}')

➡️ RETURNING возвращает строки, затронутые INSERT, UPDATE или DELETE. Идеально для логирования и цепочек операций.

🧪 Оконные функции (OVER): аналитика без костылей

Нужно пронумеровать строки, посчитать скользящее среднее или ранжировать — делай одним запросом:
# Создадим таблицу с продажами
conn.execute('''CREATE TABLE sales (
id INTEGER PRIMARY KEY,
product TEXT,
amount REAL,
sale_date DATE
)''')

# Заполняем тестовыми данными
sales_data = [
('Ноутбук', 1000, '2024-01-15'),
('Ноутбук', 1200, '2024-01-20'),
('Мышь', 50, '2024-01-10'),
('Мышь', 45, '2024-01-18'),
('Клавиатура', 80, '2024-01-12'),
]
conn.executemany('INSERT INTO sales (product, amount, sale_date) VALUES (?, ?, ?)', sales_data)

# Ранжируем товары по убыванию суммы продаж внутри каждой категории
cursor = conn.execute('''
SELECT
product,
amount,
sale_date,
ROW_NUMBER() OVER (PARTITION BY product ORDER BY amount DESC) as rank_in_product,
SUM(amount) OVER (PARTITION BY product) as total_per_product,
AVG(amount) OVER (ORDER BY sale_date ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) as moving_avg
FROM sales
ORDER BY product, amount DESC
''')

for row in cursor.fetchall():
print(row)

➡️ OVER (PARTITION BY ... ORDER BY ...) создаёт "окно" для вычислений. Не путать с GROUP BY — здесь строки не сворачиваются.

🗣 Запомни: Современный SQLite — это не просто хранилище. Используй UPSERT для атомарных обновлений, RETURNING чтобы не делать лишних запросов, а оконные функции — для аналитики прямо в БД.

Понравились скрытые фичи SQLite? Жмите 🔥
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥113
🐳 Docker для Python-разработчика (оптимизация образов)

Твой Docker-образ весит 1.5 ГБ, а сборка каждый раз тянет все зависимости с нуля? Пора это исправить. Правильный Dockerfile экономит время, трафик и нервы.

⚡️ Выбор базового образа: slim vs alpine

Главное правило: никогда не использовать python:latest. Он огромный.
#  Плохо: 1 ГБ+
FROM python:3.11

# Хорошо: ~200 МБ
FROM python:3.11-slim

# Для фанатов минимализма: ~100 МБ (но могут быть проблемы с компиляцией)
FROM python:3.11-alpine

➡️ slim — урезанный Debian, alpine — минимальный Linux на musl libc. Для большинства проектов хватает slim.

Многоступенчатая сборка (multi-stage): отделяем сборку от рантайма

Зачем тащить в продакшен компиляторы и исходники?
# Этап 1: сборка
FROM python:3.11-slim as builder

WORKDIR /app
COPY requirements.txt .
# Создаём один слой со всеми зависимостями
RUN pip install --user --no-warn-script-location -r requirements.txt

# Этап 2: финальный образ
FROM python:3.11-slim
WORKDIR /app

# Копируем только установленные пакеты из builder
COPY --from=builder /root/.local /root/.local
# Копируем код
COPY . .

# Важно: добавляем путь к пакетам
ENV PATH=/root/.local/bin:$PATH

CMD ["python", "main.py"]

➡️ В финальном образе только нужные пакеты и твой код. Компиляторы, временные файлы и кэш pip остаются в builder.

🧪 Оптимизация кэширования зависимостей: почему Docker снова качает pandas?

Docker кэширует слои. Если требования не меняются — не переустанавливай.
FROM python:3.11-slim

WORKDIR /app

# 1. Копируем ТОЛЬКО requirements.txt
COPY requirements.txt . # Этот слой кэшируется отдельно!

# 2. Устанавливаем зависимости
RUN pip install --no-cache-dir -r requirements.txt

# 3. Копируем весь остальной код
COPY . . # Этот слой пересобирается при любом изменении кода

CMD ["python", "main.py"]

➡️ Docker кэширует слой RUN pip install... до тех пор, пока requirements.txt не изменится. Изменил одну строчку в коде — зависимости не переустанавливаются.

🔍 Убираем мусор: один RUN, меньше слоев

Каждая команда в Dockerfile создаёт слой. Объединяй!
#  Плохо: 4 слоя, остаётся мусор
RUN apt-get update
RUN apt-get install -y gcc
RUN pip install pandas
RUN apt-get remove -y gcc

# Хорошо: 1 слой, чисто
RUN apt-get update && \
apt-get install -y --no-install-recommends gcc && \
pip install pandas && \
apt-get remove -y gcc && \
apt-get autoremove -y && \
rm -rf /var/lib/apt/lists/* # Чистим кэш apt

➡️ Объединяй связанные команды через &&. Флаг --no-install-recommends ставит только необходимые пакеты. rm -rf /var/lib/apt/lists/* удаляет кэш пакетов.

🚀 Практический пример: итоговый оптимизированный Dockerfile

Собираем FastAPI-приложение с зависимостями, включая компилируемые.
# Этап сборки
FROM python:3.11-slim as builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --user --no-warn-script-location -r requirements.txt

# Этап рантайма
FROM python:3.11-slim

WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY . .

ENV PATH=/root/.local/bin:$PATH \
PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1 # Отключаем .pyc файлы

# Не запускай от root!
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

➡️ Используем --no-warn-script-location чтобы не было предупреждений. PYTHONDONTWRITEBYTECODE=1 ускоряет запуск и экономит место. Запуск от не-root пользователя повышает безопасность.

🗣 Запомни: Используй slim-образы, многоступенчатую сборку и правильное кэширование зависимостей. Объединяй команды в один RUN и убирай за собой. Итоговый образ должен весить в 5-10 раз меньше первоначального.

Полезен гайд по оптимизации Docker? Жмите 🔥
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥103
🌀 Context Manager: не только `with open()`. Пишем свои для транзакций и таймеров

Контекстные менеджеры в Python — это магия with, которая управляет жизненным циклом ресурсов. Все знают про файлы, но настоящая сила открывается, когда пишешь свои менеджеры для транзакций, замеров времени и автоматического отката.

🎆 Быстрый старт: как это работает

Любой контекстный менеджер — это класс с методами __enter__ и __exit__.
class MyManager:
def __enter__(self):
print("Входим в контекст")
return self # Этот объект попадёт в as

def __exit__(self, exc_type, exc_val, exc_tb):
print("Выходим из контекста")
return False # Если True — исключение будет подавлено

with MyManager() as mgr:
print("Внутри контекста")

➡️ __enter__ выполняется при входе в with, __exit__ — всегда при выходе, даже если было исключение.

Менеджер для транзакций БД (автооткат)

Представь, что нужно гарантировать откат изменений при ошибке:
import sqlite3

class Transaction:
def __init__(self, conn):
self.conn = conn

def __enter__(self):
self.conn.execute("BEGIN")
return self.conn

def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.conn.commit()
print(" Коммит")
else:
self.conn.rollback()
print(" Откат из-за ошибки:", exc_val)
return False # Пробрасываем исключение дальше

# Использование
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE users (id INTEGER, name TEXT)")

try:
with Transaction(conn) as tx:
tx.execute("INSERT INTO users VALUES (1, 'Alice')")
raise ValueError("Искусственная ошибка!")
except:
pass

print(conn.execute("SELECT * FROM users").fetchall()) # []

➡️ Если в блоке with вылетает исключение — __exit__ делает rollback. Нет ошибки — commit.

🧪 Менеджер-таймер для замера скорости

Идеально для быстрого профайлинга кусков кода:
import time
from contextlib import contextmanager

class Timer:
def __enter__(self):
self.start = time.perf_counter()
return self

def __exit__(self, *args):
self.end = time.perf_counter()
self.elapsed = self.end - self.start
print(f"⏱️ Время выполнения: {self.elapsed:.4f} сек")

with Timer():
data = [x**2 for x in range(1_000_000)]
# ⏱️ Время выполнения: 0.0453 сек

➡️ perf_counter() даёт высокоточное время. Менеджер можно расширить, чтобы сохранять результаты в лог.

🔍 Продвинутый паттерн: менеджер с параметрами

Через __init__ передаём настройки, а в __enter__ — применяем:
class LogToFile:
def __init__(self, filename, mode="a"):
self.filename = filename
self.mode = mode

def __enter__(self):
self.file = open(self.filename, self.mode)
print(f"📂 Файл {self.filename} открыт")
return self.file

def __exit__(self, exc_type, exc_val, exc_tb):
self.file.close()
print(f"📂 Файл {self.filename} закрыт")

with LogToFile("app.log") as log:
log.write("Старт сессии\n")

➡️ Так можно оборачивать любые ресурсы: сетевые соединения, временные файлы, сессии API.

🛠 Декоратор @contextmanager из contextlib

Писать класс каждый раз долго. Есть синтаксический сахар:
from contextlib import contextmanager

@contextmanager
def temporary_file(content):
import tempfile
tmp = tempfile.NamedTemporaryFile(delete=False, mode="w")
try:
tmp.write(content)
tmp.close()
yield tmp.name # Внутри with доступно это значение
finally:
import os
os.unlink(tmp.name) # Удаляем файл после выхода

with temporary_file("секретные данные") as fpath:
print(f"Файл создан: {fpath}")
# Файл уже удалён

➡️ Всё до yield — это __enter__, всё после — __exit__. Идеально для одноразовых менеджеров.

🗣️ Запомни: Если видишь код по схеме «настрой → выполни → почисти» — это крик о помощи. Вытащи его в контекстный менеджер и живи спокойно.Если понравилось — жмите 🔥
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥65
🛡 Pydantic + typing: валидация данных на этапе разработки с помощью assert_type

Знаешь боль, когда Pydantic-модель прошла валидацию в рантайме, но mypy кричит на несоответствие типов? Решение есть — учим Pydantic и статические анализаторы понимать друг друга.

⚡️ Проблема: валидация есть, а типизации нет
from pydantic import BaseModel, field_validator
from typing import assert_type

class User(BaseModel):
id: int
name: str

@field_validator('id')
@classmethod
def validate_id(cls, v: str) -> int: # Принимает str, возвращает int
return int(v)

# Данные проходят валидацию
user = User.model_validate({"id": "123", "name": "Ivan"})
print(user.id) # 123

# Но mypy не знает, что id точно int после валидации
assert_type(user.id, int) # mypy теперь понимает

➡️ assert_type() не выполняет проверку в рантайме — это подсказка для mypy. Если типы не совпадут, анализатор ругнётся ещё до запуска.

TypeGuard: кастомная валидация с информацией для mypy

Когда стандартных валидаторов недостаточно, создаём функцию с TypeGuard:
from typing import TypeGuard
from pydantic import BaseModel

class Item(BaseModel):
value: str

def is_valid_item(data: dict) -> TypeGuard[Item]:
"""Проверяет, можно ли создать Item из data, и сообщает mypy о типе."""
try:
Item.model_validate(data)
return True
except:
return False

# Пример использования
raw_data = {"value": "test"}

if is_valid_item(raw_data):
# Здесь mypy точно знает, что raw_data подходит для Item
item = Item.model_validate(raw_data) # Без ошибок типизации
print(item.value)

➡️ TypeGuard — это явное обещание mypy: "Если функция вернула True, то тип такой-то". Работает в связке с isinstance.

🧪 Pydantic + typing.overload: точные типы для фабричных методов

Когда метод может возвращать разные модели в зависимости от входных данных:
from typing import overload, Union
from pydantic import BaseModel

class Success(BaseModel):
data: dict
status: str = "ok"

class Error(BaseModel):
error: str
status: str = "fail"

@overload
def create_response(data: dict) -> Success: ...

@overload
def create_response(data: None, error: str) -> Error: ...

def create_response(data: Union[dict, None] = None, error: str = "") -> Union[Success, Error]:
if data is not None:
return Success(data=data)
return Error(error=error)

# Теперь mypy понимает, какая модель вернётся
result1 = create_response({"user": "Ivan"})
assert_type(result1, Success) #

result2 = create_response(error="Not found")
assert_type(result2, Error) #

➡️ @overload декларативно описывает все возможные сигнатуры функции. Mypy использует это для точного вывода типов.

🔍 ConfigDict(strict=True) + typing: полный контроль

Заставляем Pydantic быть строгим, а mypy — проверять:
from pydantic import BaseModel, ConfigDict
from typing import Literal

class StrictModel(BaseModel):
model_config = ConfigDict(strict=True)

id: int
status: Literal["active", "pending"] # Только два значения

# Эта строка вызовет ошибку валидации (строка вместо int)
# user = StrictModel.model_validate({"id": "100", "status": "active"}) # ValidationError

# А mypy поймает ошибку типа ещё раньше
from typing import TYPE_CHECKING

if TYPE_CHECKING:
data: dict[str, str] = {"id": "100", "status": "active"}
# mypy: error: Argument 1 to "model_validate" of "BaseModel"
# has incompatible type "dict[str, str]"; expected "dict[str, Any]"

➡️ Включаем strict=True в конфиге, чтобы Pydantic не пытался преобразовывать типы. Mypy увидит несоответствие в TYPE_CHECKING блоке.

🗣️ Запомни: assert_type и TypeGuard — это мосты между Pydantic и mypy. Первый говорит «поверь мне, здесь такой тип», второй — «я проверил данные и гарантирую тип». Используй их, чтобы ошибки типов ловились до запуска, а не в проде.

Полезен совет по статической типизации? Жмите 🔥
Please open Telegram to view this post
VIEW IN TELEGRAM
2🔥1
🤐 Склеиваем списки с помощью `zip()`

Частая задача: есть два связанных списка (например, имена и зарплаты), и нужно пройтись по ним одновременно.

⛔️ Плохой способ (через индекс):

names = ['Anna', 'Oleg']
salaries = [100, 200]

for i in range(len(names)):
print(names[i], salaries[i])


🖥 Pythonic способ:
Функция zip берет по одному элементу из каждого списка и отдает их парами.


names = ['Anna', 'Oleg', 'Max']
salaries = [100, 200, 300]

for name, salary in zip(names, salaries):
print(f"{name} получает {salary}k")


💡 Важно: zip остановится, как только закончится самый короткий список. Если нужно сохранить все данные (заполнив пустоты None`), используйте `zip_longest из модуля itertools.

#python #bestpractices #codingtips
Please open Telegram to view this post
VIEW IN TELEGRAM
👍42
🎯 asyncio.Semaphore: ограничиваем число одновременных запросов к API

Ты делаешь 1000 асинхронных запросов к API, а сервер отвечает 429-й ошибкой? Без ограничения одновременных запросов ты рискуешь получить бан или перегрузить удаленный сервис. Semaphore - это страж, который пропускает только N корутин одновременно.

⚡️ Базовый пример: не больше 3 запросов в секунду

import asyncio
import aiohttp

async def fetch(url, session, semaphore):
    async with semaphore:  # Ждем, пока не освободится слот
        async with session.get(url) as response:
            return await response.text()

async def main():
    semaphore = asyncio.Semaphore(3)  # Не более 3 одновременных запросов
    urls = ['https://api.example.com/item/{}'.format(i) for i in range(20)]
   
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(url, session, semaphore) for url in urls]
        results = await asyncio.gather(*tasks)
    print(f'Получено {len(results)} ответов')

asyncio.run(main())

➡️ async with semaphore блокирует выполнение, если все слоты заняты. Как только один запрос завершается, слот освобождается и пускает следующий.

Semaphore как контекстный менеджер в задаче

Иногда нужно ограничить не всю функцию целиком, а только критическую секцию:

async def bounded_fetch(url, session, semaphore):
    # Тут можно делать предварительную обработку без семафора
    async with semaphore:  # только сам запрос ограничиваем
        async with session.get(url) as response:
            return await response.json()

➡️ Так ты можешь совмещать CPU-работу (которая не блокирует) и IO-вызовы с лимитом.

🔍 Продвинутое использование: разные лимиты для разных эндпоинтов

Допустим, у тебя два API: одно разрешает 5 RPS, другое - 10. Создай отдельные семафоры:

class RateLimiter:
    def __init__(self):
        self.sem_a = asyncio.Semaphore(5)
        self.sem_b = asyncio.Semaphore(10)

    async def call_api_a(self, session):
        async with self.sem_a:
            return await session.get('https://api.a.com/')

    async def call_api_b(self, session):
        async with self.sem_b:
            return await session.get('https://api.b.com/')

➡️ Изолированные лимиты для разных ресурсов.

🧪 Обработка таймаутов и повторных попыток

В реальном коде нужно быть готовым к ошибкам. Комбинируем семафор с повторными попытками:

import async_timeout

async def fetch_with_retry(url, session, semaphore, retries=3):
    for attempt in range(retries):
        try:
            async with semaphore:
                async with async_timeout.timeout(5):  # таймаут 5 секунд
                    async with session.get(url) as response:
                        return await response.text()
        except (asyncio.TimeoutError, aiohttp.ClientError) as e:
            print(f'Попытка {attempt+1} для {url} не удалась: {e}')
            if attempt == retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # экспоненциальная задержка

➡️ Если запрос провалился, пробуем снова, но не занимаем семафор во время паузы (пауза вне async with semaphore).

🚀 Практический кейс: вежливый парсер с ограничением

Допустим, парсим сайт, который разрешает 2 запроса в секунду. Используем семафор + задержку между стартами:

async def crawl(urls):
    sem = asyncio.Semaphore(2)
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in urls:
            # запускаем корутину, но не ждем её сейчас
            task = asyncio.create_task(bounded_fetch(url, session, sem))
            tasks.append(task)
            # делаем паузу между стартами, чтобы не превысить лимит даже на старте
            await asyncio.sleep(0.5)  # 2 запроса в секунду -> пауза 0.5с
        results = await asyncio.gather(*tasks)

➡️ Семафор ограничивает одновременные, а дополнительная пауза разносит старты.

🗣️ Запомни: Semaphore - это твой щит от троттлинга. Оборачивай каждый поход в сеть в async with semaphore и подбирай лимиты под API. Для разных эндпоинтов - разные семафоры. И никогда не забывай про таймауты.

Полезен совет по защите от бана? Жмите 🔥
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥62👍2
🛠 Dependency Injection в Python: зачем и как без фреймворков

Ты когда-нибудь писал класс, который внутри создаёт объекты, а потом не мог его протестировать? Или менял базу данных, а переписывать приходилось полпроекта? Dependency Injection (DI) лечит эту боль. Всё, что нужно — передавать зависимости через конструктор.

⚡️ Базовый пример: жёсткая связь vs внедрение

Плохо:
class UserService:
def __init__(self):
self.db = PostgreSQLDatabase() # Сам создаёт зависимость

def get_user(self, user_id):
return self.db.query(f"SELECT * FROM users WHERE id = {user_id}")

👉 Класс привязан к конкретной БД. Сменить на MySQL? Править код. Тестировать? Только с настоящей БД.

Хорошо:
class UserService:
def __init__(self, database): # Получает готовую зависимость
self.db = database

def get_user(self, user_id):
return self.db.query(f"SELECT * FROM users WHERE id = {user_id}")

# Теперь можно подставить любую БД с методом query
service = UserService(PostgreSQLDatabase())

👉 Класс ничего не знает о реализации. Гибкость и тестируемость растут.

✔️ Тестировать теперь легко

class MockDB:
def query(self, sql):
return {"id": 1, "name": "Тест"}

def test_get_user():
service = UserService(MockDB())
user = service.get_user(1)
assert user["name"] == "Тест"

👉 Никакой реальной БД, тесты летают.

🔍 Простой DI-контейнер своими руками

Когда зависимостей много, таскать их вручную надоедает. Сделаем простой контейнер:
class Container:
def __init__(self):
self._services = {}

def register(self, name, builder):
self._services[name] = builder

def resolve(self, name):
builder = self._services.get(name)
if not builder:
raise ValueError(f"Service {name} not found")
return builder(self)

# Настраиваем контейнер
container = Container()
container.register('db', lambda c: PostgreSQLDatabase())
container.register('user_service', lambda c: UserService(c.resolve('db')))

# Получаем готовый сервис со всеми зависимостями
user_service = container.resolve('user_service')

👉 Контейнер сам создаёт объекты, подставляя нужные зависимости. Добавил новый сервис — просто опиши его в регистрации.

🧪 Фабрики и зависимости с параметрами

Иногда объекту нужны не только сервисы, но и параметры (например, строка подключения). Контейнер легко расширить:
container.register('db', lambda c: PostgreSQLDatabase(c.resolve('config').db_url))
container.register('config', lambda c: Config.from_env())

👉 Теперь можно строить любые графы зависимостей.

🚀 Практический кейс: сервис с репозиторием и логгером
class UserRepo:
def __init__(self, db):
self.db = db
def get(self, user_id):
return self.db.query(...)

class Logger:
def log(self, msg):
print(f"[LOG] {msg}")

class UserService:
def __init__(self, repo, logger):
self.repo = repo
self.logger = logger

def get_user(self, user_id):
self.logger.log(f"Fetching user {user_id}")
return self.repo.get(user_id)

# Собираем через контейнер
container.register('db', lambda c: PostgreSQLDatabase())
container.register('repo', lambda c: UserRepo(c.resolve('db')))
container.register('logger', lambda c: Logger())
container.register('user_service', lambda c: UserService(c.resolve('repo'), c.resolve('logger')))

service = container.resolve('user_service')

👉 Все зависимости собраны в одном месте, легко подменить любую часть для тестов или замены реализации.

🗣️ Запомни: Dependency Injection — это принцип «не создай, а получи».
Please open Telegram to view this post
VIEW IN TELEGRAM
👍42
🧩 ABC vs Protocols: выбираем интерфейс правильно

В Python два способа описать контракт: абстрактные базовые классы (ABC) и протоколы. Оба говорят "у объекта должен быть метод X", но работают по-разному.

⚡️ ABC: явное наследование

from abc import ABC, abstractmethod

class Drawer(ABC):
@abstractmethod
def draw(self):
pass

class Circle(Drawer):
def draw(self):
return "🔴"

def show(obj: Drawer):
print(obj.draw())

show(Circle()) #
show("строка") # mypy ругнётся

➡️ Класс обязан наследоваться от Drawer. Жёстко, зато явно.

✔️ Protocol: структурная типизация

from typing import Protocol

class Drawer(Protocol):
def draw(self): ...

class Circle:
def draw(self):
return "🔴"

class Square:
def draw(self):
return "⬛️"

def show(obj: Drawer):
print(obj.draw())

show(Circle()) # 🔴
show(Square()) # ⬛️

➡️ Наследование не нужно. Если есть метод draw - объект подходит.

🎯 Когда что брать?

# ABC - когда нужна общая логика
class Logger(ABC):
def log(self, msg):
print(f"[{self.prefix()}] {msg}")

@abstractmethod
def prefix(self): ...

# Protocol - когда нужна гибкость
class Fetcher(Protocol):
async def fetch(self, url): ...
# Любой класс с fetch подойдёт - aiohttp, httpx, мок для тестов


🔍 Живой пример: обработчики событий

from typing import Protocol

class Handler(Protocol):
def handle(self, event: dict): ...

class UserHandler:
def handle(self, event):
print(f"Пользователь: {event}")

class OrderHandler:
def handle(self, event):
print(f"Заказ: {event}")

def process(events, handler: Handler):
for event in events:
handler.handle(event)

process([{"id": 1}], UserHandler())
process([{"order": 123}], OrderHandler()) # оба работают


🧪 Runtime-проверка если очень надо

from typing import Protocol, runtime_checkable

@runtime_checkable
class Drawer(Protocol):
def draw(self): ...

print(isinstance(Circle(), Drawer)) # True

➡️ isinstance работает, но проверяет только наличие метода, не типы аргументов.

🗣️ Запомни: ABC - для жёсткой архитектуры и общей логики. Protocol - для гибкости и утиной типизации. В современном Python протоколы используют чаще - они не заставляют лезть в наследование, если объект и так умеет что нужно.
Please open Telegram to view this post
VIEW IN TELEGRAM
2
📈 Вышел из `range()`: быстрые числовые циклы с numpy и numba

Все пишут циклы с range(), но на больших данных они тормозят как бабушкин компьютер. Для числодробилки есть два монстра: numpy (векторизация) и numba (JIT-компиляция). Разбираемся, как ускориться в десятки раз и не сойти с ума.

⚡️ Проблема: обычный цикл тормозит
import time

def sum_squares(n):
total = 0
for i in range(n):
total += i * i
return total

start = time.time()
result = sum_squares(10_000_000)
print(f"Python: {time.time() - start:.2f} сек")
# Python: 0.85 сек (примерно)

➡️ 10 миллионов итераций — почти секунда. А если нужно обработать гигабайты данных? Можно и чай сходить выпить.

Numpy: думаем векторами

Numpy выполняет операции над целыми массивами на C, без питоновских циклов.
import numpy as np

def np_sum_squares(n):
arr = np.arange(n) # массив [0, 1, 2, ..., n-1]
return np.sum(arr * arr) # всё сразу

start = time.time()
result = np_sum_squares(10_000_000)
print(f"Numpy: {time.time() - start:.2f} сек")
# Numpy: 0.05 сек (в 17 раз быстрее)

➡️ Векторизация — когда операция применяется ко всем элементам сразу. Никаких циклов, чистая скорость.

🔍 Numba: JIT-компиляция для сложной логики

А если логика не векторизуется (условия, рекурсия)? Numba компилирует питоновский код в машинный прямо перед выполнением.
from numba import njit

@njit
def nb_sum_squares(n):
total = 0
for i in range(n):
total += i * i
return total

start = time.time()
result = nb_sum_squares(10_000_000)
print(f"Numba: {time.time() - start:.2f} сек")
# Numba: 0.02 сек (в 42 раза быстрее)

➡️ Просто добавил @njit — и код полетел. Магия.

🚀 Когда что брать
# Numpy — массовые операции
arr = np.array([1, 2, 3, 4, 5])
print(arr.mean()) # 3.0
print(arr * 2) # [2 4 6 8 10]

# Numba — сложная логика с условиями
@njit
def conditional_sum(arr):
total = 0
for x in arr:
if x > 0:
total += x
return total


📦 Практический пример: обработка изображения
import numpy as np
from numba import njit

img = np.random.randint(0, 255, (1000, 1000), dtype=np.uint8)

@njit
def filter_nb(img):
h, w = img.shape
result = np.zeros((h, w), dtype=np.uint8)
for i in range(1, h-1):
for j in range(1, w-1):
val = (int(img[i-1][j]) + img[i+1][j] + img[i][j-1] + img[i][j+1]) // 4
result[i][j] = val
return result

# filter_nb(img) ~0.1 сек (в 50 раз быстрее чистого Python)

➡️ Двойной цикл по миллиону пикселей? Numba справляется играючи.

🗣️ Запомни: Если в коде есть тяжёлый цикл с числами — numpy для массовки, numba для сложной логики.
Please open Telegram to view this post
VIEW IN TELEGRAM
3👍3
🔄 State Machine (машина состояний) на match/case и классах

Когда у тебя есть объект, который может находиться в разных состояниях и менять поведение в зависимости от них (заказ, документ, чат-бот), без машины состояний код превращается в спагетти из if-else. Python даёт два удобных способа построить чистый конечный автомат: классические классы с состояниями и новый match/case.

⚡️ Проблема: лестница if-else
class Order:
def __init__(self):
self.state = "new"

def next(self):
if self.state == "new":
self.state = "paid"
print("Оплачено")
elif self.state == "paid":
self.state = "shipped"
print("Отправлено")
elif self.state == "shipped":
self.state = "delivered"
print("Доставлено")
else:
print("Уже доставлено")

➡️ Работает, но добавить новое состояние или действие — боль. Код раздувается, логика перемешана.

✔️ Решение 1: классы для каждого состояния (паттерн State)
from abc import ABC, abstractmethod

class OrderState(ABC):
@abstractmethod
def next(self, order):
pass

class NewState(OrderState):
def next(self, order):
print("Оплачено")
order.state = PaidState()

class PaidState(OrderState):
def next(self, order):
print("Отправлено")
order.state = ShippedState()

class ShippedState(OrderState):
def next(self, order):
print("Доставлено")
order.state = DeliveredState()

class DeliveredState(OrderState):
def next(self, order):
print("Заказ уже доставлен")

class Order:
def __init__(self):
self.state = NewState()

def next(self):
self.state.next(self)

➡️ Каждое состояние — отдельный класс с собственной логикой. Добавлять новые состояния легко, логика не размазана.

🔍 Решение 2: match/case (Python 3.10+)
class Order:
def __init__(self):
self.state = "new"

def next(self):
match self.state:
case "new":
self.state = "paid"
print("Оплачено")
case "paid":
self.state = "shipped"
print("Отправлено")
case "shipped":
self.state = "delivered"
print("Доставлено")
case "delivered":
print("Уже доставлено")
case _:
print("Неизвестное состояние")

➡️ match/case делает код компактнее и читаемее, но все действия по-прежнему в одном месте. Хорошо для простых автоматов.

🧪 Пример: статусы заказа с действиями

Допустим, нужно не только менять статус, но и выполнять разные действия:
from datetime import datetime

class Order:
def __init__(self):
self.state = "new"
self.history = []

def next(self):
match self.state:
case "new":
self.state = "paid"
self.history.append(("paid", datetime.now()))
print("Оплата прошла успешно")
case "paid":
self.state = "shipped"
self.history.append(("shipped", datetime.now()))
print("Заказ отправлен")
case "shipped":
self.state = "delivered"
self.history.append(("delivered", datetime.now()))
print("Заказ доставлен клиенту")
case _:
print("Нельзя перейти из текущего состояния")

➡️ К каждой смене состояния можно прицепить любую логику — запись в историю, отправку уведомлений, вызов API.

🗣️ Запомни: Машина состояний - это не rocket science. Для простоты бери match/case, для серьёзного проекта - классы. Главное - не плоди if-else на 100 строк, иначе потом сам не разберёшь, где ты и куда идёшь.
Please open Telegram to view this post
VIEW IN TELEGRAM
3👍3
🔄 Итераторы и генераторы: как for реально работает под капотом

Все пишут for x in list, но мало кто знает, что за этим скрывается протокол итератора. Разберём магию: как объекты становятся итерируемыми, как написать свой итератор и почему yield экономит память лучше любого оптимизатора.

Что на самом деле делает for

Когда ты пишешь:
items = [1, 2, 3]
for item in items:
print(item)


Python делает примерно это:
iterator = iter(items)  # получаем итератор
while True:
try:
item = next(iterator) # берём следующий элемент
print(item)
except StopIteration: # когда элементы кончились
break

➡️ Любой объект можно скормить for, если у него есть методы __iter__() или __getitem__(). Но настоящая магия — в протоколе итератора.

🔍 Протокол итератора: два метода, которые всё решают

Чтобы объект стал итерируемым, ему нужны:
- __iter__() — возвращает итератор (обычно self)
- __next__() — возвращает следующий элемент или бросает StopIteration
class Counter:
def __init__(self, limit):
self.limit = limit
self.current = 0

def __iter__(self):
return self # возвращаем себя как итератор

def __next__(self):
if self.current >= self.limit:
raise StopIteration # сигнал, что всё
self.current += 1
return self.current - 1

for x in Counter(5):
print(x) # 0 1 2 3 4

➡️ Вот так for понимает, когда остановиться. Всё честно, без магии.

Свой итератор для бесконечной последовательности

Хочешь бесконечный счётчик? Легко:
class InfiniteCounter:
def __init__(self, start=0):
self.current = start

def __iter__(self):
return self

def __next__(self):
self.current += 1
return self.current - 1

counter = InfiniteCounter()
for i in counter:
print(i)
if i > 5:
break
# 0 1 2 3 4 5

➡️ Бесконечность не предел. Главное — не забыть break, иначе уйдёшь в космос.

🧪 Проблема: писать классы каждый раз лень

Согласись, тащить целый класс ради одного итератора — жирно. Тут на сцену выходят генераторы.
def counter_gen(limit):
for i in range(limit):
yield i

for x in counter_gen(5):
print(x) # 0 1 2 3 4

➡️ yield превращает функцию в генератор. При каждом вызове next() она выполняется до следующего yield и замораживает состояние.

🚀 Бесконечный генератор в одну строку
def infinite_gen(start=0):
while True:
yield start
start += 1

gen = infinite_gen()
print(next(gen)) # 0
print(next(gen)) # 1
print(next(gen)) # 2

➡️ Памяти занимает ровно столько, сколько нужно для хранения текущего состояния. Никаких гигантских списков.

📦 Почему генераторы экономят память: наглядный пример
import sys

# Список
numbers_list = [x for x in range(1_000_000)]
print(sys.getsizeof(numbers_list)) # ~8 MB

# Генератор
numbers_gen = (x for x in range(1_000_000))
print(sys.getsizeof(numbers_gen)) # ~112 bytes

➡️ Список хранит все миллион элементов в памяти. Генератор — только текущее состояние и ссылку на функцию. Разница в тысячи раз.

🔧 Реальный кейс: читаем гигабайтный файл
def read_large_file(file_path):
with open(file_path, 'r') as f:
for line in f:
yield line.strip()

# Используем
for line in read_large_file("huge_log.txt"):
if "ERROR" in line:
print(line)

➡️ Файл читается построчно, в памяти только одна строка. Можно обрабатывать терабайты логов на ноутбуке с 4GB RAM.

⚡️ Генераторные выражения: list comprehension без памяти
# Список (жадный)
squares_list = [x*x for x in range(1000)]

# Генератор (ленивый)
squares_gen = (x*x for x in range(1000))

print(sum(squares_gen)) # можно посчитать сумму без создания списка

➡️ Круглые скобки вместо квадратных — и миллион элементов не жрут память.

🗣️ Запомни: for - это просто синтаксический сахар над iter() и next() с ловлей StopIteration. Генераторы с yield позволяют писать ленивые последовательности без классов.
Please open Telegram to view this post
VIEW IN TELEGRAM
6👍3🔥2🤷‍♀1
Media is too big
VIEW IN TELEGRAM
👩‍💻 Уроки Python с нуля / #17 – Основы ООП. Создание класса и объекта

В этом видео автор приступает к одной из важнейших тем в Python — объектно-ориентированному программированию. Вы узнаете теоретическую основу ООП и научитесь на практике создавать свои первые классы и объекты в языке Питон, закладывая фундамент для написания сложных и структурированных программ.


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

🗣️ Запомни: класс — это шаблон или чертеж, а объект — конкретный экземпляр, созданный по этому чертежу. ООП помогает управлять сложностью кода.

🤩 Pytstart || #Видеокурс
Please open Telegram to view this post
VIEW IN TELEGRAM
4🔥1
Media is too big
VIEW IN TELEGRAM
👩‍💻 Уроки Python с нуля – Наследование, инкапсуляция, полиморфизм

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


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

🗣️Запомни: наследование, инкапсуляция и полиморфизм — это три кита ООП. Освоив их, вы перестанете писать "процедурный" код и начнёте проектировать архитектуру по-настоящему сложных систем.

🤩 Pytstart || #Видеокурс
Please open Telegram to view this post
VIEW IN TELEGRAM
❤‍🔥43🔥1