{
"title": "15 промтов для генерации Python кода: от скриптов до FastAPI",
"content": "## Введение\n\nГенерация кода с помощью AI — уже не эксперимент, а рабочий инструмент каждого Python-разработчика. По данным опроса Stack Overflow 2025 года, более 40% респондентов используют AI-ассистенты для написания кода ежедневно. Правильно составленный промт позволяет не просто получить шаблон, а сразу рабочий, хорошо структурированный код с учётом лучших практик. В этой подборке — 15 проверенных промтов, которые я сам использую в продакшене: от однострочных скриптов до полноценного FastAPI-приложения с тестами. Каждый промт сопровождается реальным примером вывода и пояснением, где это может пригодиться.\n\n> Важно: Все промты ориентированы на современный Python (3.11+), используют typing и следуют PEP 8. Результаты сгенерированы Claude 3.5 Sonnet (актуальная версия на июль 2026).\n\n---\n\n## 1. Базовая функция с type hints и документацией\n\nПромт:\n> Напиши на Python функцию `parse_log_line(line: str) -> dict[str, str
| None], которая парсит строку лога формата[2026-07-25 10:00:00] ERROR: Module 'foo' - Connection refused. Верни словарь с ключамиtimestamp,level,module,message. Добавь docstring с примерами и обработку неправильного формата черезtry/except.\n\n**Пример использования:**\n- Парсинг больших текстовых файлов логов\n- Интеграция с системами логирования\n\n**Сгенерированный код:**\n```python\nimport re\nfrom datetime import datetime\n\ndef parse_log_line(line: str) -> dict[str, str | None]:\n \"\"\"\n Парсит строку лога формата [timestamp] LEVEL: Module - Message.\n\n Примеры:\n >>> parse_log_line('[2026-07-25 10:00:00] ERROR: Module foo - Connection refused')\n {'timestamp': '2026-07-25 10:00:00', 'level': 'ERROR', 'module': 'foo', 'message': 'Connection refused'}\n\n >>> parse_log_line('invalid line') is None\n True\n\n Args:\n line: Строка лога.\n\n Returns:\n Словарь с полями или None при ошибке.\n \"\"\"\n pattern = r'^\\[(\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2})\\]\\s+(\\w+):\\s+(.*?)\\s+-\\s+(.*)$'\n try:\n match = re.match(pattern, line.strip())\n if not match:\n return None\n timestamp_str, level, module, message = match.groups()\n # Валидация timestamp\n datetime.strptime(timestamp_str, '%Y-%m-%d %H:%M:%S')\n return {'timestamp': timestamp_str, 'level': level, 'module': module, 'message': message}\n except (ValueError, AttributeError):\n return None\n```\n\n**Почему это работает:** Промт чётко задаёт сигнатуру, формат входных данных и требования к обработке ошибок. AI возвращает код, готовый к копированию с минимальными правками. Обратите внимание на использованиеdatetime.strptimeдля валидации — это часто упускают из виду.\n\n---\n\n## 2. Генератор CSV-отчёта из списка словарей\n\n**Промт:**\n> Напиши Python-скрипт, который принимает список словарейdata(например, с ключамиname,age,email) и записывает CSV-файл с заголовками. Используйcsv.DictWriter. Обработай случай, если словари имеют разный набор ключей — заполняй пропуски пустой строкой. Добавь аннотации типов.\n\n**Пример использования:**\n- Выгрузка данных из API в CSV\n- Экспорт результатов тестирования\n\n**Сгенерированный код:**\n```python\nimport csv\nfrom typing import Any\n\ndef write_csv(filename: str, data: list[dict[str, Any]]) -> None:\n \"\"\"\n Записывает список словарей в CSV, автоматически определяя заголовки.\n Если ключи различаются, недостающие поля заполняются пустой строкой.\n \"\"\"\n if not data:\n raise ValueError(\"Data list is empty\")\n # Собираем все возможные ключи из всех словарей\n fieldnames: list[str] = []\n for row in data:\n for key in row:\n if key not in fieldnames:\n fieldnames.append(key)\n\n with open(filename, 'w', newline='', encoding='utf-8') as f:\n writer = csv.DictWriter(f, fieldnames=fieldnames, restval='')\n writer.writeheader()\n writer.writerows(data)\n```\n\n**Совет:** Используйтеrestval=''для заполнения отсутствующих полей — это стандартная фичаcsv.DictWriter, о которой забывают даже опытные разработчики.\n\n---\n\n## 3. Веб-скрапинг с BeautifulSoup4 и защитой от блокировки\n\n**Промт:**\n> Напиши функцию на Pythonfetch_article_text(url: str) -> str, которая загружает HTML-страницу, используяrequestsс User-Agent реального браузера (Mozilla/5.0...), и извлекает весь текст из тегов
и
-. Добавь обработку ошибок (статус не 200, таймаут 10 сек) и задержку 1-2 секунды черезtime.sleep(random.uniform(1,2)). Верни очищенный текст без лишних пробелов.\n\n**Пример использования:**\n- Сбор статей для NLP-задач\n- Мониторинг цен\n\n**Сгенерированный код:**\n```python\nimport time\nimport random\nimport requests\nfrom bs4 import BeautifulSoup\n\ndef fetch_article_text(url: str) -> str | None:\n headers = {\n 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'\n }\n try:\n response = requests.get(url, headers=headers, timeout=10)\n response.raise_for_status()\n time.sleep(random.uniform(1, 2))\n soup = BeautifulSoup(response.text, 'html.parser')\n texts = []\n for tag in soup.find_all(['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6']):\n texts.append(tag.get_text(strip=True))\n return '\\n\\n'.join(texts)\n except (requests.RequestException, AttributeError) as e:\n print(f\"Error fetching {url}: {e}\")\n return None\n```\n\n**Важно:** Реальные User-Agent — обязательное условие для промышленного скрапинга. Без этого многие сайты возвращают 403. Источник: [HTTP User-Agent best practices](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/User-Agent).\n\n---\n\n## 4. Асинхронный загрузчик с asyncio и aiohttp\n\n**Промт:**\n> Напиши асинхронную функциюasync def download_urls(urls: list[str]) -> dict[str, str], которая принимает список URL и возвращает словарь{url: content}. Используйaiohttp.ClientSessionс лимитом в 5 одновременных соединений (TCPConnector(limit=5)). Обработай ошибки соединения и timeout (10 сек). Верни пустую строку для неудачных запросов.\n\n**Пример использования:**\n- Быстрая загрузка множества API-эндпоинтов\n- Парсинг тысяч страниц\n\n**Сгенерированный код:**\n```python\nimport asyncio\nimport aiohttp\nfrom typing import Any\n\nasync def download_urls(urls: list[str]) -> dict[str, str]:\n results: dict[str, str] = {}\n connector = aiohttp.TCPConnector(limit=5)\n async with aiohttp.ClientSession(connector=connector) as session:\n async def fetch(url: str) -> None:\n try:\n async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:\n results[url] = await resp.text()\n except Exception:\n results[url] = ''\n tasks = [fetch(url) for url in urls]\n await asyncio.gather(*tasks, return_exceptions=True)\n return results\n```\n\n**Практический совет:**return_exceptions=Trueпредотвращает падение всего сценария из-за одной плохой ссылки. Это стандартный паттерн при параллельной загрузке.\n\n---\n\n## 5. FastAPI: CRUD для списка задач (in-memory)\n\n**Промт:**\n> Напиши FastAPI-приложение с эндпоинтами:GET /tasks— список всех задач,POST /tasks— создание задачи (обязательно поляtitle: str,done: bool = False, возвращает задачу с id),GET /tasks/{id}— получение задачи,PUT /tasks/{id}— обновление (done),DELETE /tasks/{id}. Используй Pydantic-моделиTaskCreate,Task(с id и датой созданияcreated_atтипаdatetime). Храни данные в глобальном спискеtasks: list[Task]. Добавь обработку 404. В ответе используйresponse_model.\n\n**Пример использования:**\n- Прототип REST API за 5 минут\n- Основа для тестового задания\n\n**Сгенерированный код:**\n```python\nfrom fastapi import FastAPI, HTTPException\nfrom pydantic import BaseModel\nfrom datetime import datetime\nfrom typing import Optional\nimport uuid\n\napp = FastAPI()\n\nclass TaskCreate(BaseModel):\n title: str\n done: bool = False\n\nclass Task(TaskCreate):\n id: str\n created_at: datetime\n\n class Config:\n from_attributes = True\n\ntasks: list[Task] = []\n\n@app.get(\"/tasks\", response_model=list[Task])\ndef list_tasks():\n return tasks\n\n@app.post(\"/tasks\", response_model=Task, status_code=201)\ndef create_task(task: TaskCreate):\n new_task = Task(id=str(uuid.uuid4()), created_at=datetime.utcnow(), **task.dict())\n tasks.append(new_task)\n return new_task\n\n@app.get(\"/tasks/{task_id}\", response_model=Task)\ndef get_task(task_id: str):\n for t in tasks:\n if t.id == task_id:\n return t\n raise HTTPException(status_code=404, detail=\"Task not found\")\n\n@app.put(\"/tasks/{task_id}\", response_model=Task)\ndef update_task(task_id: str, update: TaskCreate):\n for t in tasks:\n if t.id == task_id:\n t.title = update.title\n t.done = update.done\n return t\n raise HTTPException(status_code=404, detail=\"Task not found\")\n\n@app.delete(\"/tasks/{task_id}\", status_code=204)\ndef delete_task(task_id: str):\n for i, t in enumerate(tasks):\n if t.id == task_id:\n del tasks[i]\n return\n raise HTTPException(status_code=404, detail=\"Task not found\")\n```\n\n**Источник:** Официальная документация FastAPI по моделям Pydantic — [pydantic-docs.helpmanual.io](https://pydantic-docs.helpmanual.io/).\n\n---\n\n## 6. FastAPI с SQLAlchemy и PostgreSQL\n\n**Промт:**\n> Напиши FastAPI-приложение с подключением к PostgreSQL через SQLAlchemy (асинхронный режим). Создай модельUserс полямиid(int, PK),email(unique),name(str). Реализуй эндпоинты:GET /users— список,POST /users— создание. Используйasync defиasync with db.session. Для конфигурации используйDATABASE_URLиз переменной окружения. Добавь Alembic-миграцию.\n\n**Пример использования:**\n- Боевой проект с БД\n- Микросервис с пользователями\n\n**Сгенерированный код (ключевые фрагменты):**\n```python\n# config.py\nfrom pydantic_settings import BaseSettings\n\nclass Settings(BaseSettings):\n DATABASE_URL: str = \"postgresql+asyncpg://user:pass@localhost/db\"\n\nsettings = Settings()\n\n# database.py\nfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker\n\nengine = create_async_engine(settings.DATABASE_URL, echo=True)\nasync_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)\n\nasync def get_db() -> AsyncSession:\n async with async_session() as session:\n yield session\n\n# models.py\nfrom sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column\n\nclass Base(DeclarativeBase):\n pass\n\nclass User(Base):\n __tablename__ = \"users\"\n id: Mapped[int] = mapped_column(primary_key=True)\n email: Mapped[str] = mapped_column(unique=True)\n name: Mapped[str]\n```\n\n**Важно:** В промте явно указан асинхронный режим иasync def— без этого AI часто генерирует синхронный код. Также упомянуты переменные окружения (12-factor app).\n\n---\n\n## 7. Юнит-тесты с pytest и фикстурами\n\n**Промт:**\n> Напиши набор pytest-тестов для функцииcalc_stats(numbers: list[float]) -> dict[str, float], которая возвращаетmean,median,std,min,max. Используй фикстуруsample_data(например, [1,2,3,4,5]). Проверь, чтоmean = 3,median = 3,std = ~1.581. Добавь параметризованный тест с граничными случаями: пустой список, список с одним элементом, список с отрицательными числами. Используйpytest.approxдля сравнения float.\n\n**Пример использования:**\n- Гарантия корректности вычислений\n- CI/CD пайплайн\n\n**Сгенерированный код:**\n```python\nimport pytest\nfrom math import sqrt\n\ndef calc_stats(numbers: list[float]) -> dict[str, float]:\n n = len(numbers)\n if n == 0:\n return {'mean': 0.0, 'median': 0.0, 'std': 0.0, 'min': 0.0, 'max': 0.0}\n mean = sum(numbers) / n\n sorted_nums = sorted(numbers)\n mid = n // 2\n if n % 2 == 0:\n median = (sorted_nums[mid - 1] + sorted_nums[mid]) / 2\n else:\n median = sorted_nums[mid]\n variance = sum((x - mean) ** 2 for x in numbers) / n\n std = sqrt(variance)\n return {'mean': mean, 'median': median, 'std': std, 'min': min(numbers), 'max': max(numbers)}\n\n# tests.py\n@pytest.fixture\ndef sample_data():\n return [1.0, 2.0, 3.0, 4.0, 5.0]\n\ndef test_calc_stats_normal(sample_data):\n result = calc_stats(sample_data)\n assert result['mean'] == 3.0\n assert result['median'] == 3.0\n assert result['std'] == pytest.approx(1.5811388300841898, rel=1e-3)\n assert result['min'] == 1.0\n assert result['max'] == 5.0\n\n@pytest.mark.parametrize(\"data,expected_mean,expected_median\", [\n ([], 0.0, 0.0),\n ([42.0], 42.0, 42.0),\n ([-5.0, 0.0, 5.0], 0.0, 0.0),\n])\ndef test_calc_stats_edge_cases(data, expected_mean, expected_median):\n result = calc_stats(data)\n assert result['mean'] == pytest.approx(expected_mean)\n assert result['median'] == pytest.approx(expected_median)\n```\n\n**Зачем это:** Тесты на граничные случаи — 80% багов в продакшене возникает именно на них. Pytest — де-факто стандарт в Python-сообществе (документация: [docs.pytest.org](https://docs.pytest.org/)).\n\n---\n\n## 8. Работа с JSON: фильтрация и агрегация\n\n**Промт:**\n> Напиши функциюfilter_orders(orders: list[dict], date_from: str, date_to: str, min_total: float = 0) -> list[dict], которая принимает список заказов (каждый с полямиorder_id,date,total,items), оставляет только те, гдеdateмеждуdate_fromиdate_to(формат YYYY-MM-DD) иtotal >= min_total. Результат отсортируй поtotalпо убыванию. Используйdatetimeдля сравнения, а не строки. Пропусти невалидные строки дат.\n\n**Пример использования:**\n- Фильтрация данных из CRM\n- Аналитика продаж\n\n**Сгенерированный код:**\n```python\nfrom datetime import datetime\nfrom typing import Any\n\ndef filter_orders(orders: list[dict[str, Any]], date_from: str, date_to: str, min_total: float = 0) -> list[dict[str, Any]]:\n try:\n dt_from = datetime.strptime(date_from, '%Y-%m-%d')\n dt_to = datetime.strptime(date_to, '%Y-%m-%d')\n except ValueError:\n raise ValueError(\"date_from and date_to must be in YYYY-MM-DD format\")\n\n filtered = []\n for order in orders:\n try:\n order_date = datetime.strptime(order.get('date', ''), '%Y-%m-%d')\n except (ValueError, KeyError):\n continue\n if dt_from <= order_date <= dt_to and order.get('total', 0) >= min_total:\n filtered.append(order)\n filtered.sort(key=lambda x: x.get('total', 0), reverse=True)\n return filtered\n```\n\n**Нюанс:** Сравнение строк с датами в формате YYYY-MM-DD корректно работает лексикографически, но лучше преобразовывать вdatetime— так код устойчив к изменению формата. Это известная рекомендация из [PEP 8](https://peps.python.org/pep-0008/).\n\n---\n\n## 9. Декоратор для логирования времени выполнения\n\n**Промт:**\n> Напиши декоратор@log_execution_timeс параметромunit: str = 's'(секунды/миллисекунды), который выводит вprintимя функции и время её выполнения. Используйtime.perf_counter. Декторатор должен корректно обрабатыватьasyncфункции. Если функция асинхронная — оборачивай вasync def.\n\n**Пример использования:**\n- Мониторинг производительности\n- Профилирование запросов\n\n**Сгенерированный код:**\n```python\nimport time\nimport functools\nimport asyncio\nfrom typing import Callable, TypeVar, Any\n\nF = TypeVar('F', bound=Callable[..., Any])\n\ndef log_execution_time(unit: str = 's') -> Callable[[F], F]:\n def decorator(func: F) -> F:\n if asyncio.iscoroutinefunction(func):\n @functools.wraps(func)\n async def async_wrapper(*args: Any, **kwargs: Any) -> Any:\n start = time.perf_counter()\n result = await func(*args, **kwargs)\n elapsed = time.perf_counter() - start\n if unit == 'ms':\n print(f\"{func.__name__} took {elapsed*1000:.2f} ms\")\n else:\n print(f\"{func.__name__} took {elapsed:.4f} s\")\n return result\n return async_wrapper # type: ignore\n else:\n @functools.wraps(func)\n def sync_wrapper(*args: Any, **kwargs: Any) -> Any:\n start = time.perf_counter()\n result = func(*args, **kwargs)\n elapsed = time.perf_counter() - start\n if unit == 'ms':\n print(f\"{func.__name__} took {elapsed*1000:.2f} ms\")\n else:\n print(f\"{func.__name__} took {elapsed:.4f} s\")\n return result\n return sync_wrapper # type: ignore\n return decorator\n```\n\n**Практическое замечание:**time.perf_counterдаёт наилучшую точность на разных платформах, в отличие отtime.time. Источник: [документация Python](https://docs.python.org/3/library/time.html#time.perf_counter).\n\n---\n\n## 10. Генератор тестовых данных с Faker\n\n**Промт:**\n> Напиши функциюgenerate_users(count: int = 100) -> list[dict], которая используетfakerдля создания списка пользователей с полями:full_name,email(уникальный),phone,address,birth_date(дата от 1960 до 2005),salary(float от 30000 до 150000). Гарантируй, что email уникальны. Верни готовый список.\n\n**Пример использования:**\n- Наполнение тестовой базы\n- Демо-данные для презентаций\n\n**Сгенерированный код:**\n```python\nfrom faker import Faker\nimport random\nfrom typing import Any\n\nfake = Faker()\n\ndef generate_users(count: int = 100) -> list[dict[str, Any]]:\n emails = set()\n users = []\n while len(users) < count:\n email = fake.unique.email()\n if email in emails:\n continue\n emails.add(email)\n user = {\n 'full_name': fake.name(),\n 'email': email,\n 'phone': fake.phone_number(),\n 'address': fake.address().replace('\\n', ', '),\n 'birth_date': fake.date_of_birth(minimum_age=18, maximum_age=65).isoformat(),\n 'salary': round(random.uniform(30000, 150000), 2)\n }\n users.append(user)\n return users\n```\n\n**Важно:**fake.uniqueиспользует внутренний пул для уникальности, но может выбросить исключение, если кончились варианты. Для 100 пользователей это безопасно. Подробнее: [Faker docs](https://faker.readthedocs.io/en/master/providers.html).\n\n---\n\n## 11. Асинхронный парсер командной строки с argparse\n\n**Промт:**\n> Напиши скрипт на Python, который принимает аргументы командной строки:--input(обязательный, путь к файлу),--output(опциональный, по умолчанию stdout),--verbose(флаг). Скрипт читает JSON-файл, сортирует его содержимое (предполагается массив чисел или строк) и выводит в файл или консоль. Используйargparseиasyncio/aiofilesдля асинхронного чтения. Ошибки выводи в stderr.\n\n**Пример использования:**\n- Утилиты для обработки данных\n- ETL-конвейеры\n\n**Сгенерированный код (фрагмент):**\n```python\nimport argparse\nimport asyncio\nimport aiofiles\nimport json\nimport sys\n\nasync def main():\n parser = argparse.ArgumentParser(description='Sort JSON array')\n parser.add_argument('--input', required=True, help='Path to input JSON file')\n parser.add_argument('--output', default=None, help='Path to output file (default: stdout)')\n parser.add_argument('--verbose', action='store_true', help='Verbose output')\n args = parser.parse_args()\n\n try:\n async with aiofiles.open(args.input, 'r') as f:\n content = await f.read()\n data = json.loads(content)\n if not isinstance(data, list):\n raise TypeError(\"JSON must be an array\")\n sorted_data = sorted(data)\n output = json.dumps(sorted_data, indent=2)\n if args.output:\n async with aiofiles.open(args.output, 'w') as f:\n await f.write(output)\n else:\n print(output)\n except Exception as e:\n print(f\"Error: {e}\", file=sys.stderr)\n sys.exit(1)\n\nif __name__ == '__main__':\n asyncio.run(main())\n```\n\n**Совет:** Всегда используйтеasyncio.run()в точке входа — это гарантирует правильное управление циклом событий.\n\n---\n\n## 12. Dataclasses для конфигурации с валидацией\n\n**Промт:**\n> Создай dataclassAppConfigс полями:debug: bool = False,port: int = 8000(валидация 1024-65535),allowed_hosts: list[str],db_url: str. Добавь методfrom_dict(cls, data: dict)для загрузки из словаря с валидацией черезpost_init. Еслиportвне диапазона — выбросьValueError. Используйfieldсdefault_factoryдля списка.\n\n**Пример использования:**\n- Загрузка конфигурации из YAML/JSON\n- Type-safe конфиги\n\n**Сгенерированный код:**\n```python\nfrom dataclasses import dataclass, field\nfrom typing import Any\n\n@dataclass\nclass AppConfig:\n debug: bool = False\n port: int = 8000\n allowed_hosts: list[str] = field(default_factory=list)\n db_url: str = ''\n\n def __post_init__(self):\n if not 1024 <= self.port <= 65535:\n raise ValueError(f\"Port {self.port} out of range (1024-65535)\")\n \n @classmethod\n def from_dict(cls, data: dict[str, Any]) -> 'AppConfig':\n return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})\n```\n\n**Почему dataclasses:** Они заменяют ~70% ручного boilerplate, а валидация черезpost_init— простой и эффективный способ контроля данных. Подробнее: [PEP 557](https://peps.python.org/pep-0557/).\n\n---\n\n## 13. Работа с environment переменными через pydantic-settings\n\n**Промт:**\n> Напиши классSettingsна основеpydantic_settings.BaseSettings, который читает переменные окружения с префиксомAPP_. Поля:database_url(str),secret_key(str, min_length=16),max_connections(int, default=10),log_level(str, default=\"INFO\"). Добавь валидациюsecret_keyчерез@field_validator— если короче 16 символов, кидай ошибку. ИспользуйConfigDict(env_prefix='APP_').\n\n**Пример использования:**\n- 12-factor app конфигурация\n- Безопасное хранение секретов\n\n**Сгенерированный код:**\n```python\nfrom pydantic_settings import BaseSettings, SettingsConfigDict\nfrom pydantic import field_validator\n\nclass Settings(BaseSettings):\n model_config = SettingsConfigDict(env_prefix='APP_', env_file='.env')\n\n database_url: str\n secret_key: str\n max_connections: int = 10\n log_level: str = 'INFO'\n\n @field_validator('secret_key')\n @classmethod\n def check_min_length(cls, v: str) -> str:\n if len(v) < 16:\n raise ValueError('secret_key must be at least 16 characters')\n return v\n```\n\n**Источник:** [pydantic-settings documentation](https://docs.pydantic.dev/latest/concepts/pydantic_settings/).\n\n---\n\n## 14. Асинхронный воркер очереди с asyncio.Queue\n\n**Промт:**\n> Напиши async-воркер, который обрабатывает задачи из очереди. Определи классTask(dataclass) с полямиid(int),payload(str). Реализуй функциюworker(name: str, queue: asyncio.Queue, results: dict), которая бесконечно извлекает задачи, обрабатывает имитируя работу (await asyncio.sleep(0.5)), и сохраняет результат (uppercase payload) в словарь results. Главная корутина запускает 3 воркера и помещает 10 задач в очередь. Выведи итоговые результаты.\n\n**Пример использования:**\n- Фоновая обработка в веб-приложениях\n- Пакетная обработка данных\n\n**Сгенерированный код:**\n```python\nimport asyncio\nfrom dataclasses import dataclass\nfrom typing import Any\n\n@dataclass\nclass Task:\n id: int\n payload: str\n\nasync def worker(name: str, queue: asyncio.Queue[Task], results: dict[int, str]):\n while True:\n task = await queue.get()\n print(f\"Worker {name} processing task {task.id}\")\n await asyncio.sleep(0.5) # simulate work\n results[task.id] = task.payload.upper()\n queue.task_done()\n\nasync def main():\n queue = asyncio.Queue()\n results: dict[int, str] = {}\n \n workers = [asyncio.create_task(worker(f\"W{i}\", queue, results)) for i in range(3)]\n \n for i in range(10):\n await queue.put(Task(id=i, payload=f\"item_{i}\"))\n \n await queue.join()\n for w in workers:\n w.cancel()\n print(\"Results:\", results)\n\nif __name__ == '__main__':\n asyncio.run(main())\n```\n\n**Важный приём:** Использованиеqueue.task_done()иawait queue.join()— правильный способ завершения, без него программа завершится до обработки всех задач.\n\n---\n\n## 15. Генерация документации с помощью рефлексии\n\n**Промт:**\n> Напиши функциюgenerate_api_docs(module) -> str, которая импортирует указанный модуль и генерирует Markdown-документацию: для каждой функции (исключая приватные с_`) выводит имя, docstring (первые 3 строки), список аннотированных парамет
. Добавь обработку ошибок (статус не 200, таймаут 10 сек) и задержку 1-2 секунды черезtime.sleep(random.uniform(1,2)). Верни очищенный текст без лишних пробелов.\n\n**Пример использования:**\n- Сбор статей для NLP-задач\n- Мониторинг цен\n\n**Сгенерированный код:**\n```python\nimport time\nimport random\nimport requests\nfrom bs4 import BeautifulSoup\n\ndef fetch_article_text(url: str) -> str | None:\n headers = {\n 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'\n }\n try:\n response = requests.get(url, headers=headers, timeout=10)\n response.raise_for_status()\n time.sleep(random.uniform(1, 2))\n soup = BeautifulSoup(response.text, 'html.parser')\n texts = []\n for tag in soup.find_all(['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6']):\n texts.append(tag.get_text(strip=True))\n return '\\n\\n'.join(texts)\n except (requests.RequestException, AttributeError) as e:\n print(f\"Error fetching {url}: {e}\")\n return None\n```\n\n**Важно:** Реальные User-Agent — обязательное условие для промышленного скрапинга. Без этого многие сайты возвращают 403. Источник: [HTTP User-Agent best practices](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/User-Agent).\n\n---\n\n## 4. Асинхронный загрузчик с asyncio и aiohttp\n\n**Промт:**\n> Напиши асинхронную функциюasync def download_urls(urls: list[str]) -> dict[str, str], которая принимает список URL и возвращает словарь{url: content}. Используйaiohttp.ClientSessionс лимитом в 5 одновременных соединений (TCPConnector(limit=5)). Обработай ошибки соединения и timeout (10 сек). Верни пустую строку для неудачных запросов.\n\n**Пример использования:**\n- Быстрая загрузка множества API-эндпоинтов\n- Парсинг тысяч страниц\n\n**Сгенерированный код:**\n```python\nimport asyncio\nimport aiohttp\nfrom typing import Any\n\nasync def download_urls(urls: list[str]) -> dict[str, str]:\n results: dict[str, str] = {}\n connector = aiohttp.TCPConnector(limit=5)\n async with aiohttp.ClientSession(connector=connector) as session:\n async def fetch(url: str) -> None:\n try:\n async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:\n results[url] = await resp.text()\n except Exception:\n results[url] = ''\n tasks = [fetch(url) for url in urls]\n await asyncio.gather(*tasks, return_exceptions=True)\n return results\n```\n\n**Практический совет:**return_exceptions=Trueпредотвращает падение всего сценария из-за одной плохой ссылки. Это стандартный паттерн при параллельной загрузке.\n\n---\n\n## 5. FastAPI: CRUD для списка задач (in-memory)\n\n**Промт:**\n> Напиши FastAPI-приложение с эндпоинтами:GET /tasks— список всех задач,POST /tasks— создание задачи (обязательно поляtitle: str,done: bool = False, возвращает задачу с id),GET /tasks/{id}— получение задачи,PUT /tasks/{id}— обновление (done),DELETE /tasks/{id}. Используй Pydantic-моделиTaskCreate,Task(с id и датой созданияcreated_atтипаdatetime). Храни данные в глобальном спискеtasks: list[Task]. Добавь обработку 404. В ответе используйresponse_model.\n\n**Пример использования:**\n- Прототип REST API за 5 минут\n- Основа для тестового задания\n\n**Сгенерированный код:**\n```python\nfrom fastapi import FastAPI, HTTPException\nfrom pydantic import BaseModel\nfrom datetime import datetime\nfrom typing import Optional\nimport uuid\n\napp = FastAPI()\n\nclass TaskCreate(BaseModel):\n title: str\n done: bool = False\n\nclass Task(TaskCreate):\n id: str\n created_at: datetime\n\n class Config:\n from_attributes = True\n\ntasks: list[Task] = []\n\n@app.get(\"/tasks\", response_model=list[Task])\ndef list_tasks():\n return tasks\n\n@app.post(\"/tasks\", response_model=Task, status_code=201)\ndef create_task(task: TaskCreate):\n new_task = Task(id=str(uuid.uuid4()), created_at=datetime.utcnow(), **task.dict())\n tasks.append(new_task)\n return new_task\n\n@app.get(\"/tasks/{task_id}\", response_model=Task)\ndef get_task(task_id: str):\n for t in tasks:\n if t.id == task_id:\n return t\n raise HTTPException(status_code=404, detail=\"Task not found\")\n\n@app.put(\"/tasks/{task_id}\", response_model=Task)\ndef update_task(task_id: str, update: TaskCreate):\n for t in tasks:\n if t.id == task_id:\n t.title = update.title\n t.done = update.done\n return t\n raise HTTPException(status_code=404, detail=\"Task not found\")\n\n@app.delete(\"/tasks/{task_id}\", status_code=204)\ndef delete_task(task_id: str):\n for i, t in enumerate(tasks):\n if t.id == task_id:\n del tasks[i]\n return\n raise HTTPException(status_code=404, detail=\"Task not found\")\n```\n\n**Источник:** Официальная документация FastAPI по моделям Pydantic — [pydantic-docs.helpmanual.io](https://pydantic-docs.helpmanual.io/).\n\n---\n\n## 6. FastAPI с SQLAlchemy и PostgreSQL\n\n**Промт:**\n> Напиши FastAPI-приложение с подключением к PostgreSQL через SQLAlchemy (асинхронный режим). Создай модельUserс полямиid(int, PK),email(unique),name(str). Реализуй эндпоинты:GET /users— список,POST /users— создание. Используйasync defиasync with db.session. Для конфигурации используйDATABASE_URLиз переменной окружения. Добавь Alembic-миграцию.\n\n**Пример использования:**\n- Боевой проект с БД\n- Микросервис с пользователями\n\n**Сгенерированный код (ключевые фрагменты):**\n```python\n# config.py\nfrom pydantic_settings import BaseSettings\n\nclass Settings(BaseSettings):\n DATABASE_URL: str = \"postgresql+asyncpg://user:pass@localhost/db\"\n\nsettings = Settings()\n\n# database.py\nfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker\n\nengine = create_async_engine(settings.DATABASE_URL, echo=True)\nasync_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)\n\nasync def get_db() -> AsyncSession:\n async with async_session() as session:\n yield session\n\n# models.py\nfrom sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column\n\nclass Base(DeclarativeBase):\n pass\n\nclass User(Base):\n __tablename__ = \"users\"\n id: Mapped[int] = mapped_column(primary_key=True)\n email: Mapped[str] = mapped_column(unique=True)\n name: Mapped[str]\n```\n\n**Важно:** В промте явно указан асинхронный режим иasync def— без этого AI часто генерирует синхронный код. Также упомянуты переменные окружения (12-factor app).\n\n---\n\n## 7. Юнит-тесты с pytest и фикстурами\n\n**Промт:**\n> Напиши набор pytest-тестов для функцииcalc_stats(numbers: list[float]) -> dict[str, float], которая возвращаетmean,median,std,min,max. Используй фикстуруsample_data(например, [1,2,3,4,5]). Проверь, чтоmean = 3,median = 3,std = ~1.581. Добавь параметризованный тест с граничными случаями: пустой список, список с одним элементом, список с отрицательными числами. Используйpytest.approxдля сравнения float.\n\n**Пример использования:**\n- Гарантия корректности вычислений\n- CI/CD пайплайн\n\n**Сгенерированный код:**\n```python\nimport pytest\nfrom math import sqrt\n\ndef calc_stats(numbers: list[float]) -> dict[str, float]:\n n = len(numbers)\n if n == 0:\n return {'mean': 0.0, 'median': 0.0, 'std': 0.0, 'min': 0.0, 'max': 0.0}\n mean = sum(numbers) / n\n sorted_nums = sorted(numbers)\n mid = n // 2\n if n % 2 == 0:\n median = (sorted_nums[mid - 1] + sorted_nums[mid]) / 2\n else:\n median = sorted_nums[mid]\n variance = sum((x - mean) ** 2 for x in numbers) / n\n std = sqrt(variance)\n return {'mean': mean, 'median': median, 'std': std, 'min': min(numbers), 'max': max(numbers)}\n\n# tests.py\n@pytest.fixture\ndef sample_data():\n return [1.0, 2.0, 3.0, 4.0, 5.0]\n\ndef test_calc_stats_normal(sample_data):\n result = calc_stats(sample_data)\n assert result['mean'] == 3.0\n assert result['median'] == 3.0\n assert result['std'] == pytest.approx(1.5811388300841898, rel=1e-3)\n assert result['min'] == 1.0\n assert result['max'] == 5.0\n\n@pytest.mark.parametrize(\"data,expected_mean,expected_median\", [\n ([], 0.0, 0.0),\n ([42.0], 42.0, 42.0),\n ([-5.0, 0.0, 5.0], 0.0, 0.0),\n])\ndef test_calc_stats_edge_cases(data, expected_mean, expected_median):\n result = calc_stats(data)\n assert result['mean'] == pytest.approx(expected_mean)\n assert result['median'] == pytest.approx(expected_median)\n```\n\n**Зачем это:** Тесты на граничные случаи — 80% багов в продакшене возникает именно на них. Pytest — де-факто стандарт в Python-сообществе (документация: [docs.pytest.org](https://docs.pytest.org/)).\n\n---\n\n## 8. Работа с JSON: фильтрация и агрегация\n\n**Промт:**\n> Напиши функциюfilter_orders(orders: list[dict], date_from: str, date_to: str, min_total: float = 0) -> list[dict], которая принимает список заказов (каждый с полямиorder_id,date,total,items), оставляет только те, гдеdateмеждуdate_fromиdate_to(формат YYYY-MM-DD) иtotal >= min_total. Результат отсортируй поtotalпо убыванию. Используйdatetimeдля сравнения, а не строки. Пропусти невалидные строки дат.\n\n**Пример использования:**\n- Фильтрация данных из CRM\n- Аналитика продаж\n\n**Сгенерированный код:**\n```python\nfrom datetime import datetime\nfrom typing import Any\n\ndef filter_orders(orders: list[dict[str, Any]], date_from: str, date_to: str, min_total: float = 0) -> list[dict[str, Any]]:\n try:\n dt_from = datetime.strptime(date_from, '%Y-%m-%d')\n dt_to = datetime.strptime(date_to, '%Y-%m-%d')\n except ValueError:\n raise ValueError(\"date_from and date_to must be in YYYY-MM-DD format\")\n\n filtered = []\n for order in orders:\n try:\n order_date = datetime.strptime(order.get('date', ''), '%Y-%m-%d')\n except (ValueError, KeyError):\n continue\n if dt_from <= order_date <= dt_to and order.get('total', 0) >= min_total:\n filtered.append(order)\n filtered.sort(key=lambda x: x.get('total', 0), reverse=True)\n return filtered\n```\n\n**Нюанс:** Сравнение строк с датами в формате YYYY-MM-DD корректно работает лексикографически, но лучше преобразовывать вdatetime— так код устойчив к изменению формата. Это известная рекомендация из [PEP 8](https://peps.python.org/pep-0008/).\n\n---\n\n## 9. Декоратор для логирования времени выполнения\n\n**Промт:**\n> Напиши декоратор@log_execution_timeс параметромunit: str = 's'(секунды/миллисекунды), который выводит вprintимя функции и время её выполнения. Используйtime.perf_counter. Декторатор должен корректно обрабатыватьasyncфункции. Если функция асинхронная — оборачивай вasync def.\n\n**Пример использования:**\n- Мониторинг производительности\n- Профилирование запросов\n\n**Сгенерированный код:**\n```python\nimport time\nimport functools\nimport asyncio\nfrom typing import Callable, TypeVar, Any\n\nF = TypeVar('F', bound=Callable[..., Any])\n\ndef log_execution_time(unit: str = 's') -> Callable[[F], F]:\n def decorator(func: F) -> F:\n if asyncio.iscoroutinefunction(func):\n @functools.wraps(func)\n async def async_wrapper(*args: Any, **kwargs: Any) -> Any:\n start = time.perf_counter()\n result = await func(*args, **kwargs)\n elapsed = time.perf_counter() - start\n if unit == 'ms':\n print(f\"{func.__name__} took {elapsed*1000:.2f} ms\")\n else:\n print(f\"{func.__name__} took {elapsed:.4f} s\")\n return result\n return async_wrapper # type: ignore\n else:\n @functools.wraps(func)\n def sync_wrapper(*args: Any, **kwargs: Any) -> Any:\n start = time.perf_counter()\n result = func(*args, **kwargs)\n elapsed = time.perf_counter() - start\n if unit == 'ms':\n print(f\"{func.__name__} took {elapsed*1000:.2f} ms\")\n else:\n print(f\"{func.__name__} took {elapsed:.4f} s\")\n return result\n return sync_wrapper # type: ignore\n return decorator\n```\n\n**Практическое замечание:**time.perf_counterдаёт наилучшую точность на разных платформах, в отличие отtime.time. Источник: [документация Python](https://docs.python.org/3/library/time.html#time.perf_counter).\n\n---\n\n## 10. Генератор тестовых данных с Faker\n\n**Промт:**\n> Напиши функциюgenerate_users(count: int = 100) -> list[dict], которая используетfakerдля создания списка пользователей с полями:full_name,email(уникальный),phone,address,birth_date(дата от 1960 до 2005),salary(float от 30000 до 150000). Гарантируй, что email уникальны. Верни готовый список.\n\n**Пример использования:**\n- Наполнение тестовой базы\n- Демо-данные для презентаций\n\n**Сгенерированный код:**\n```python\nfrom faker import Faker\nimport random\nfrom typing import Any\n\nfake = Faker()\n\ndef generate_users(count: int = 100) -> list[dict[str, Any]]:\n emails = set()\n users = []\n while len(users) < count:\n email = fake.unique.email()\n if email in emails:\n continue\n emails.add(email)\n user = {\n 'full_name': fake.name(),\n 'email': email,\n 'phone': fake.phone_number(),\n 'address': fake.address().replace('\\n', ', '),\n 'birth_date': fake.date_of_birth(minimum_age=18, maximum_age=65).isoformat(),\n 'salary': round(random.uniform(30000, 150000), 2)\n }\n users.append(user)\n return users\n```\n\n**Важно:**fake.uniqueиспользует внутренний пул для уникальности, но может выбросить исключение, если кончились варианты. Для 100 пользователей это безопасно. Подробнее: [Faker docs](https://faker.readthedocs.io/en/master/providers.html).\n\n---\n\n## 11. Асинхронный парсер командной строки с argparse\n\n**Промт:**\n> Напиши скрипт на Python, который принимает аргументы командной строки:--input(обязательный, путь к файлу),--output(опциональный, по умолчанию stdout),--verbose(флаг). Скрипт читает JSON-файл, сортирует его содержимое (предполагается массив чисел или строк) и выводит в файл или консоль. Используйargparseиasyncio/aiofilesдля асинхронного чтения. Ошибки выводи в stderr.\n\n**Пример использования:**\n- Утилиты для обработки данных\n- ETL-конвейеры\n\n**Сгенерированный код (фрагмент):**\n```python\nimport argparse\nimport asyncio\nimport aiofiles\nimport json\nimport sys\n\nasync def main():\n parser = argparse.ArgumentParser(description='Sort JSON array')\n parser.add_argument('--input', required=True, help='Path to input JSON file')\n parser.add_argument('--output', default=None, help='Path to output file (default: stdout)')\n parser.add_argument('--verbose', action='store_true', help='Verbose output')\n args = parser.parse_args()\n\n try:\n async with aiofiles.open(args.input, 'r') as f:\n content = await f.read()\n data = json.loads(content)\n if not isinstance(data, list):\n raise TypeError(\"JSON must be an array\")\n sorted_data = sorted(data)\n output = json.dumps(sorted_data, indent=2)\n if args.output:\n async with aiofiles.open(args.output, 'w') as f:\n await f.write(output)\n else:\n print(output)\n except Exception as e:\n print(f\"Error: {e}\", file=sys.stderr)\n sys.exit(1)\n\nif __name__ == '__main__':\n asyncio.run(main())\n```\n\n**Совет:** Всегда используйтеasyncio.run()в точке входа — это гарантирует правильное управление циклом событий.\n\n---\n\n## 12. Dataclasses для конфигурации с валидацией\n\n**Промт:**\n> Создай dataclassAppConfigс полями:debug: bool = False,port: int = 8000(валидация 1024-65535),allowed_hosts: list[str],db_url: str. Добавь методfrom_dict(cls, data: dict)для загрузки из словаря с валидацией черезpost_init. Еслиportвне диапазона — выбросьValueError. Используйfieldсdefault_factoryдля списка.\n\n**Пример использования:**\n- Загрузка конфигурации из YAML/JSON\n- Type-safe конфиги\n\n**Сгенерированный код:**\n```python\nfrom dataclasses import dataclass, field\nfrom typing import Any\n\n@dataclass\nclass AppConfig:\n debug: bool = False\n port: int = 8000\n allowed_hosts: list[str] = field(default_factory=list)\n db_url: str = ''\n\n def __post_init__(self):\n if not 1024 <= self.port <= 65535:\n raise ValueError(f\"Port {self.port} out of range (1024-65535)\")\n \n @classmethod\n def from_dict(cls, data: dict[str, Any]) -> 'AppConfig':\n return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})\n```\n\n**Почему dataclasses:** Они заменяют ~70% ручного boilerplate, а валидация черезpost_init— простой и эффективный способ контроля данных. Подробнее: [PEP 557](https://peps.python.org/pep-0557/).\n\n---\n\n## 13. Работа с environment переменными через pydantic-settings\n\n**Промт:**\n> Напиши классSettingsна основеpydantic_settings.BaseSettings, который читает переменные окружения с префиксомAPP_. Поля:database_url(str),secret_key(str, min_length=16),max_connections(int, default=10),log_level(str, default=\"INFO\"). Добавь валидациюsecret_keyчерез@field_validator— если короче 16 символов, кидай ошибку. ИспользуйConfigDict(env_prefix='APP_').\n\n**Пример использования:**\n- 12-factor app конфигурация\n- Безопасное хранение секретов\n\n**Сгенерированный код:**\n```python\nfrom pydantic_settings import BaseSettings, SettingsConfigDict\nfrom pydantic import field_validator\n\nclass Settings(BaseSettings):\n model_config = SettingsConfigDict(env_prefix='APP_', env_file='.env')\n\n database_url: str\n secret_key: str\n max_connections: int = 10\n log_level: str = 'INFO'\n\n @field_validator('secret_key')\n @classmethod\n def check_min_length(cls, v: str) -> str:\n if len(v) < 16:\n raise ValueError('secret_key must be at least 16 characters')\n return v\n```\n\n**Источник:** [pydantic-settings documentation](https://docs.pydantic.dev/latest/concepts/pydantic_settings/).\n\n---\n\n## 14. Асинхронный воркер очереди с asyncio.Queue\n\n**Промт:**\n> Напиши async-воркер, который обрабатывает задачи из очереди. Определи классTask(dataclass) с полямиid(int),payload(str). Реализуй функциюworker(name: str, queue: asyncio.Queue, results: dict), которая бесконечно извлекает задачи, обрабатывает имитируя работу (await asyncio.sleep(0.5)), и сохраняет результат (uppercase payload) в словарь results. Главная корутина запускает 3 воркера и помещает 10 задач в очередь. Выведи итоговые результаты.\n\n**Пример использования:**\n- Фоновая обработка в веб-приложениях\n- Пакетная обработка данных\n\n**Сгенерированный код:**\n```python\nimport asyncio\nfrom dataclasses import dataclass\nfrom typing import Any\n\n@dataclass\nclass Task:\n id: int\n payload: str\n\nasync def worker(name: str, queue: asyncio.Queue[Task], results: dict[int, str]):\n while True:\n task = await queue.get()\n print(f\"Worker {name} processing task {task.id}\")\n await asyncio.sleep(0.5) # simulate work\n results[task.id] = task.payload.upper()\n queue.task_done()\n\nasync def main():\n queue = asyncio.Queue()\n results: dict[int, str] = {}\n \n workers = [asyncio.create_task(worker(f\"W{i}\", queue, results)) for i in range(3)]\n \n for i in range(10):\n await queue.put(Task(id=i, payload=f\"item_{i}\"))\n \n await queue.join()\n for w in workers:\n w.cancel()\n print(\"Results:\", results)\n\nif __name__ == '__main__':\n asyncio.run(main())\n```\n\n**Важный приём:** Использованиеqueue.task_done()иawait queue.join()— правильный способ завершения, без него программа завершится до обработки всех задач.\n\n---\n\n## 15. Генерация документации с помощью рефлексии\n\n**Промт:**\n> Напиши функциюgenerate_api_docs(module) -> str, которая импортирует указанный модуль и генерирует Markdown-документацию: для каждой функции (исключая приватные с_`) выводит имя, docstring (первые 3 строки), список аннотированных парамет
Комментарии