
- Главная
- Каталог
- Интернет технологии
- PytStart | Программирование на Python
PytStart | Программирование на Python
Канал о разработке приложений на Python, в том числе написание backend и web-приложений. Аудитория канала - начинающие или опытные Python программисты.
Статистика канала
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 сек (примерно){}
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 раз быстрее){}
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){}
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. Жёстко, зато явно.
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()) # оба работают{}
from typing import Protocol, runtime_checkable
@runtime_checkable
class Drawer(Protocol):
def draw(self): ...
print(isinstance(Circle(), Drawer)) # True{}
isinstance работает, но проверяет только наличие метода, не типы аргументов.
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}"){}
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"] == "Тест"{}
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'){}
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 блокирует выполнение, если все слоты заняты. Как только один запрос завершается, слот освобождается и пускает следующий.
async def bounded_fetch(url, session, semaphore):
# Тут можно делать предварительную обработку без семафора
async with semaphore: # только сам запрос ограничиваем
async with session.get(url) as response:
return await response.json(){}
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).
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. Для разных эндпоинтов - разные семафоры. И никогда не забывай про таймауты.
Полезен совет по защите от бана? Жмите 🔥
names = ['Anna', 'Oleg']
salaries = [100, 200]
for i in range(len(names)):
print(names[i], salaries[i])
{}
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 #codingtipsassert_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:
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.
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 использует это для точного вывода типов.
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. Первый говорит «поверь мне, здесь такой тип», второй — «я проверил данные и гарантирую тип». Используй их, чтобы ошибки типов ловились до запуска, а не в проде.
Полезен совет по статической типизации? Жмите 🔥Отзывы канала
всего 4 отзыва
- Добавлен: Сначала новые
- Добавлен: Сначала старые
- Оценка: По убыванию
- Оценка: По возрастанию
Каталог Телеграм-каналов для нативных размещений
PytStart | Программирование на Python — это Telegam канал в категории «Интернет технологии», который предлагает эффективные форматы для размещения рекламных постов в Телеграмме. Количество подписчиков канала в 4.5K и качественный контент помогают брендам привлекать внимание аудитории и увеличивать охват. Рейтинг канала составляет 22.7, количество отзывов – 4, со средней оценкой 5.0.
Вы можете запустить рекламную кампанию через сервис Telega.in, выбрав удобный формат размещения. Платформа обеспечивает прозрачные условия сотрудничества и предоставляет детальную аналитику. Стоимость размещения составляет 1398.6 ₽, а за 10 выполненных заявок канал зарекомендовал себя как надежный партнер для рекламы в TG. Размещайте интеграции уже сегодня и привлекайте новых клиентов вместе с Telega.in!
Вы снова сможете добавить каналы в корзину из каталога
Комментарий