Примеры кода на Python 2026: лучшие практики, разбор и задачи с решениями

# установка uv (Linux/macOS)
curl -LsSf https://astral.sh/uv/install.sh | sh

# новый проект
uv init my-project
cd my-project

# добавить зависимости
uv add httpx pydantic
uv add --dev pytest ruff mypy

# запуск
uv run main.py
uv run pytest

# проверки
uv run ruff check . && uv run ruff format .
uvx ty check                    # быстрый тайпчекер (бета)

# нужная версия Python, в том числе free-threaded сборка
uv python install 3.14 3.14t

[project]
name = "my-project"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["httpx", "pydantic"]

[tool.ruff]
line-length = 100

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM"]

result = []
for i in range(len(users)):
    if users[i]["active"] == True:
        result.append(users[i]["name"].upper())

active_names = [user["name"].upper() for user in users if user["active"]]

# enumerate вместо счётчика
for index, item in enumerate(items, start=1):
    print(f"{index}. {item}")

# zip для параллельного обхода (strict=True ловит разную длину)
for name, score in zip(names, scores, strict=True):
    print(f"{name}: {score}")

# распаковка
first, *middle, last = [1, 2, 3, 4, 5]

# словарь со значением по умолчанию
counts = {}
for word in words:
    counts[word] = counts.get(word, 0) + 1

# ещё лучше — Counter
from collections import Counter
top = Counter(words).most_common(3)

# walrus-оператор :=
if (n := len(data)) > 100:
    print(f"Слишком много данных: {n}")

# объединение словарей
config = defaults | user_settings

from collections.abc import Iterable, Callable

type UserId = int
type JSON = dict[str, "JSON"] | list["JSON"] | str | int | float | bool | None


def first[T](items: Iterable[T], default: T | None = None) -> T | None:
    """Вернуть первый элемент или значение по умолчанию."""
    return next(iter(items), default)


class Stack[T]:
    def __init__(self) -> None:
        self._items: list[T] = []

    def push(self, item: T) -> None:
        self._items.append(item)

    def pop(self) -> T:
        if not self._items:
            raise IndexError("стек пуст")
        return self._items.pop()


def apply_twice[T](func: Callable[[T], T], value: T) -> T:
    return func(func(value))

from dataclasses import dataclass, field


@dataclass(slots=True, frozen=True)
class Product:
    name: str
    price: float
    tags: list[str] = field(default_factory=list)

    @property
    def price_with_vat(self) -> float:
        return round(self.price * 1.2, 2)


p = Product("Клавиатура", 3500)
print(p.price_with_vat)  # 4200.0

from pydantic import BaseModel, EmailStr, Field


class UserIn(BaseModel):
    name: str = Field(min_length=2, max_length=50)
    email: EmailStr
    age: int = Field(ge=0, le=150)


user = UserIn.model_validate({"name": "Анна", "email": "anna@example.com", "age": 28})
print(user.model_dump_json())

def handle(event: dict) -> str:
    match event:
        case {"type": "click", "x": int(x), "y": int(y)}:
            return f"Клик в точке ({x}, {y})"
        case {"type": "key", "key": "Enter" | "Return"}:
            return "Нажат Enter"
        case {"type": "key", "key": str(key)}:
            return f"Клавиша {key}"
        case {"type": t}:
            return f"Неизвестное событие: {t}"
        case _:
            return "Некорректные данные"


print(handle({"type": "click", "x": 10, "y": 20}))

import asyncio
import httpx

URLS = [
    "https://www.python.org",
    "https://docs.python.org/3/",
    "https://pypi.org",
]


async def fetch(client: httpx.AsyncClient, url: str) -> tuple[str, int]:
    response = await client.get(url, follow_redirects=True)
    return url, response.status_code


async def main() -> None:
    async with httpx.AsyncClient(timeout=10) as client:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch(client, url)) for url in URLS]

    for task in tasks:
        url, status = task.result()
        print(f"{status} — {url}")


if __name__ == "__main__":
    asyncio.run(main())

sem = asyncio.Semaphore(5)

async def fetch_limited(client: httpx.AsyncClient, url: str):
    async with sem:
        return await fetch(client, url)

async with asyncio.timeout(5):
    await long_operation()

import os
path = os.path.join(os.getcwd(), "data", "report.txt")
f = open(path)
text = f.read()
f.close()

from pathlib import Path

path = Path.cwd() / "data" / "report.txt"
text = path.read_text(encoding="utf-8")

# все CSV-файлы рекурсивно
for csv_file in Path("data").rglob("*.csv"):
    print(csv_file.name, csv_file.stat().st_size)

# создать папку, если её нет
Path("output").mkdir(parents=True, exist_ok=True)

class ValidationError(Exception):
    """Ошибка валидации пользовательских данных."""


def parse_age(raw: str) -> int:
    try:
        age = int(raw)
    except ValueError as exc:
        raise ValidationError(f"Возраст должен быть числом, получено: {raw!r}") from exc
    if not 0 <= age <= 150:
        raise ValidationError(f"Недопустимый возраст: {age}")
    return age

try:
    raise ExceptionGroup("ошибки", [ValueError("a"), TypeError("b")])
except* ValueError as group:
    print("ValueError:", group.exceptions)
except* TypeError as group:
    print("TypeError:", group.exceptions)

import time
from contextlib import contextmanager


@contextmanager
def timer(label: str):
    start = time.perf_counter()
    try:
        yield
    finally:
        print(f"{label}: {time.perf_counter() - start:.3f} с")


with timer("Сортировка"):
    sorted(range(1_000_000), reverse=True)

import functools
import time


def retry(times: int = 3, delay: float = 1.0, exceptions=(Exception,)):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as exc:
                    if attempt == times:
                        raise
                    print(f"Попытка {attempt} не удалась: {exc}. Повтор через {delay} с")
                    time.sleep(delay)
        return wrapper
    return decorator


@retry(times=3, delay=0.5, exceptions=(ConnectionError,))
def unstable_request():
    ...

from functools import cache


@cache
def fib(n: int) -> int:
    return n if n < 2 else fib(n - 1) + fib(n - 2)


print(fib(100))

from pathlib import Path


def read_large_file(path: Path):
    with path.open(encoding="utf-8") as f:
        for line in f:
            yield line.rstrip("\n")


errors = (line for line in read_large_file(Path("app.log")) if "ERROR" in line)
for line in errors:
    print(line)

from itertools import batched, chain, groupby, islice, pairwise

list(batched(range(10), 3))        # [(0,1,2), (3,4,5), (6,7,8), (9,)]  — Python 3.12+
list(pairwise([1, 2, 3, 4]))       # [(1,2), (2,3), (3,4)]
list(chain([1, 2], [3], [4, 5]))   # [1, 2, 3, 4, 5]
list(islice(range(100), 5))        # [0, 1, 2, 3, 4]

data = sorted(["apple", "avocado", "banana", "cherry"], key=lambda s: s[0])
for letter, group in groupby(data, key=lambda s: s[0]):
    print(letter, list(group))

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
)
logger = logging.getLogger(__name__)


def process(order_id: int) -> None:
    logger.info("Обработка заказа %s", order_id)
    try:
        ...
    except Exception:
        logger.exception("Ошибка при обработке заказа %s", order_id)
        raise

git clone https://github.com/justxor/python-primery-koda-2026.git
cd python-primery-koda-2026
uv run python -m advanced.asyncio_patterns   # запустить любой пример
uv run --with pytest pytest                  # прогнать все тесты

class Positive:
    def __set_name__(self, owner, name):        # Python сам сообщит имя атрибута
        self.name, self.private = name, f"_{name}"

    def __get__(self, instance, owner=None):
        return self if instance is None else getattr(instance, self.private)

    def __set__(self, instance, value):
        if not isinstance(value, (int, float)) or value <= 0:
            raise ValueError(f"{self.name} должно быть > 0, получено {value!r}")
        setattr(instance, self.private, value)


class Product:
    price = Positive()
    quantity = Positive()

    def __init__(self, price, quantity):
        self.price = price          # вызывает Positive.__set__
        self.quantity = quantity


Product(100, 2)
Product(-1, 2)   # ValueError: price должно быть > 0, получено -1

class Exporter:
    registry: dict[str, type["Exporter"]] = {}

    def __init_subclass__(cls, /, fmt: str, **kwargs):
        super().__init_subclass__(**kwargs)
        Exporter.registry[fmt] = cls


class JsonExporter(Exporter, fmt="json"): ...
class CsvExporter(Exporter, fmt="csv"): ...

print(Exporter.registry)   # {'json': <class JsonExporter>, 'csv': <class CsvExporter>}

class SingletonMeta(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]


class Settings(metaclass=SingletonMeta): ...

assert Settings() is Settings()

from collections.abc import Callable
from typing import Protocol, Self
import functools

type JSON = dict[str, JSON] | list[JSON] | str | int | float | bool | None   # рекурсивный алиас


class Stack[T]:                            # дженерик-класс без TypeVar
    def __init__(self) -> None:
        self._items: list[T] = []

    def push(self, item: T) -> Self:       # Self — для цепочек вызовов
        self._items.append(item)
        return self


class Comparable(Protocol):                # структурная типизация
    def __lt__(self, other: Self, /) -> bool: ...


def max_item[C: Comparable](items: list[C]) -> C:
    return max(items)


def logged[**P, R](func: Callable[P, R]) -> Callable[P, R]:   # ParamSpec
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        print("вызов", func.__name__)
        return func(*args, **kwargs)
    return wrapper

import asyncio


async def gather_limited(coros, limit: int):
    sem = asyncio.Semaphore(limit)

    async def run(coro):
        async with sem:
            return await coro

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(run(c)) for c in coros]
    return [t.result() for t in tasks]

import random


async def retry_async(func, *, attempts=5, base_delay=0.1, max_delay=5.0):
    for attempt in range(attempts):
        try:
            return await func()
        except (ConnectionError, TimeoutError):
            if attempt == attempts - 1:
                raise
            await asyncio.sleep(random.uniform(0, min(max_delay, base_delay * 2**attempt)))

import sys, sysconfig
import concurrent.futures as cf

print("free-threaded:", bool(sysconfig.get_config_var("Py_GIL_DISABLED")))
print("GIL включён:", sys._is_gil_enabled())      # Python 3.13+


def best_executor() -> type[cf.Executor]:
    if not sys._is_gil_enabled():
        return cf.ThreadPoolExecutor
    if hasattr(cf, "InterpreterPoolExecutor"):    # Python 3.14+
        return cf.InterpreterPoolExecutor
    return cf.ProcessPoolExecutor

from contextvars import ContextVar
import logging

request_id: ContextVar[str] = ContextVar("request_id", default="-")


class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id.get()   # id запроса в каждой строке лога
        return True


async def handle(rid: str):
    token = request_id.set(rid)
    try:
        await do_work()        # глубоко внутри request_id.get() вернёт rid
    finally:
        request_id.reset(token)

from dataclasses import dataclass


@dataclass(slots=True, frozen=True)
class Point:
    x: float
    y: float
# экземпляр без __dict__: ~48 байт вместо ~340, доступ к атрибутам быстрее

python -m cProfile -s cumulative script.py | head -20
python -X importtime -c "import mymodule"     # что медленно импортируется

def running_average():
    total = count = 0
    average = 0.0
    while True:
        value = yield average          # получаем значение через send()
        total += value
        count += 1
        average = total / count


avg = running_average()
next(avg)                              # «прогрев» до первого yield
avg.send(10); avg.send(20)             # 15.0

errors = count_by("service", only("ERROR", parse(open("app.log"))))

from contextlib import ExitStack

with ExitStack() as stack:
    files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths]
    ...   # все файлы гарантированно закроются

from dataclasses import dataclass


@dataclass(frozen=True)
class Ok[T]:
    value: T


@dataclass(frozen=True)
class Err[E]:
    error: E


type Result[T, E] = Ok[T] | Err[E]


def safe_divide(a: float, b: float) -> Result[float, str]:
    return Err("деление на ноль") if b == 0 else Ok(a / b)


match safe_divide(1, 0):
    case Ok(value):
        print("результат", value)
    case Err(error):
        print("ошибка", error)

from string.templatelib import Interpolation, Template


def sql(template: Template) -> tuple[str, list]:
    query, params = [], []
    for part in template:
        if isinstance(part, Interpolation):
            query.append("?")
            params.append(part.value)
        else:
            query.append(part)
    return "".join(query), params


name = "Robert'); DROP TABLE students;--"
sql(t"SELECT * FROM users WHERE name = {name}")
# ('SELECT * FROM users WHERE name = ?', ["Robert'); DROP TABLE students;--"])

from string.templatelib import Template

name = "<script>alert(1)</script>"
template: Template = t"Привет, {name}!"

print(template.strings)                     # ('Привет, ', '!')
print([i.value for i in template.interpolations])  # ['<script>alert(1)</script>']

try:
    connect()
except TimeoutError, ConnectionRefusedError:
    print("Сервер недоступен")

lazy import json                      # json загрузится только при первом использовании
lazy from pathlib import Path

lists = [[1, 2], [3, 4], [5]]
print([*L for L in lists])            # [1, 2, 3, 4, 5]

dicts = [{"a": 1}, {"b": 2}, {"a": 3}]
print({**d for d in dicts})           # {'a': 3, 'b': 2}

config = frozendict(host="localhost", port=8080)   # неизменяемый и хешируемый
cache = {config: "подключение"}                    # можно использовать как ключ

MISSING = sentinel("MISSING")         # вместо object() — с понятным repr


def get(key: str, default=MISSING):
    if default is MISSING:
        ...

python -m profiling.sampling attach 12345
python -m profiling.sampling run --flamegraph script.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="Заметки API")


class NoteIn(BaseModel):
    title: str
    text: str = ""


class Note(NoteIn):
    id: int


notes: dict[int, Note] = {}


@app.post("/notes", response_model=Note, status_code=201)
def create_note(data: NoteIn) -> Note:
    note = Note(id=len(notes) + 1, **data.model_dump())
    notes[note.id] = note
    return note


@app.get("/notes/{note_id}", response_model=Note)
def get_note(note_id: int) -> Note:
    if note_id not in notes:
        raise HTTPException(status_code=404, detail="Заметка не найдена")
    return notes[note_id]

# test_utils.py
import pytest
from utils import is_palindrome, is_balanced


@pytest.mark.parametrize(
    ("text", "expected"),
    [
        ("шалаш", True),
        ("А роза упала на лапу Азора", True),
        ("python", False),
        ("", True),
    ],
)
def test_is_palindrome(text: str, expected: bool) -> None:
    assert is_palindrome(text) is expected


def test_is_balanced_empty_string() -> None:
    assert is_balanced("")


@pytest.fixture
def sample_users() -> list[dict]:
    return [{"name": "Анна", "active": True}, {"name": "Борис", "active": False}]


def test_active_users(sample_users: list[dict]) -> None:
    active = [u["name"] for u in sample_users if u["active"]]
    assert active == ["Анна"]

python-primery-koda-2026/
├── README.md                    # руководство: примеры, задачи, вопросы
├── advanced/                    # продвинутые темы: код, который можно запустить
│   ├── descriptors.py           # 11. дескрипторы
│   ├── metaprogramming.py       # 12. __init_subclass__, метаклассы
│   ├── typing_advanced.py       # 13. PEP 695, Protocol, ParamSpec
│   ├── asyncio_patterns.py      # 14. паттерны asyncio
│   ├── concurrency.py           # 15. GIL, free-threading
│   ├── contextvars_demo.py      # 16. contextvars
│   ├── performance.py           # 17. производительность и память
│   ├── generators_advanced.py   # 18. генераторы, ExitStack
│   ├── patterns.py              # 19. Result, DI, Circuit Breaker
│   ├── python314.py             # 20. t-строки, annotationlib
│   └── tasks_senior.py          # решения задач 8–13
├── tests/                       # pytest-тесты для всего кода
├── pyproject.toml
└── .github/workflows/tests.yml  # CI: Python 3.12, 3.13, 3.14