Создавал ли собственные пакеты (Python packages) для распространения кода?

Да, создавал. Например, я разработал пакет etl_helpers для стандартизации ETL-задач в нашей команде. Он содержал общие connectors к источникам данных, шаблоны для инкрементальной загрузки и утилиты для логирования и обработки ошибок.

Процесс создания и ключевые решения:

1. Структура проекта: Использовал структуру src/, чтобы избежать путаницы между пакетом и рабочим каталогом. Основные модули: connectors, incremental, monitoring.
2. Управление зависимостями: В pyproject.toml чётко разделил зависимости на dependencies (например, pandas, sqlalchemy) и dev-dependencies (pytest, black, mypy).
3. Тестирование: Настроил pytest с фикстурами для тестирования connectors на mock-серверах (например, pytest-httpserver).
4. Сборка и публикация: Автоматизировал сборку через python -m build и публикацию в наш приватный PyPi-репозиторий (Artifactory) с помощью GitHub Actions.

Пример содержимого pyproject.toml:

[project]
name = "etl-helpers"
version = "1.2.0"
authors = [{name = "Your Name", email = "name@example.com"}]
description = "Internal ETL utilities for data pipelines."
readme = "README.md"
requires-python = ">=3.9"
dependencies = [
    "pandas>=1.5",
    "sqlalchemy>=2.0",
    "psycopg2-binary",
    "requests",
]

[project.optional-dependencies]
dev = ["pytest", "black", "mypy", "pytest-httpserver"]

[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

Пример кода из модуля инкрементальной загрузки:

# src/etl_helpers/incremental.py
from datetime import datetime
from typing import Any, Dict
import hashlib

class IncrementalLoader:
    """Класс для управления состоянием инкрементальной загрузки."""
    def __init__(self, state_storage: Dict[str, Any]):
        self.state = state_storage

    def get_last_processed_id(self, table_name: str) -> int:
        """Возвращает последний обработанный ID для таблицы."""
        return self.state.get(f"last_id_{table_name}", 0)

    def update_state(self, table_name: str, last_id: int, checksum: str = None):
        """Обновляет состояние после успешной загрузки."""
        self.state[f"last_id_{table_name}"] = last_id
        self.state[f"updated_at_{table_name}"] = datetime.utcnow().isoformat()
        if checksum:
            self.state[f"checksum_{table_name}"] = checksum

Пакет позволил унифицировать код в разных пайплайнах и упростил онбординг новых разработчиков.