Как создать свою библиотеку на Python и зачем это может быть нужно?

Как создать свою библиотеку на Python и зачем это может быть нужно

Зачем создавать собственную библиотеку

1. Переиспользование кода между проектами

Если вы пишете одну и ту же логику в разных проектах, лучше вынести её в библиотеку:

# Без библиотеки: копируешь функцию в каждый проект
# data_extraction.py в Project A
def extract_from_api(endpoint, auth_key):
    response = requests.get(endpoint, headers={"X-API-Key": auth_key})
    return response.json()

# data_extraction.py в Project B
def extract_from_api(endpoint, auth_key):  # Дублирование!
    response = requests.get(endpoint, headers={"X-API-Key": auth_key})
    return response.json()

# С библиотекой: импортируешь
from mycompany_datatools.extractors import extract_from_api

2. Стандартизация кода в компании

Все проекты используют одинаковый способ логирования, обработки ошибок, работы с БД.

3. Упрощение сложной логики

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

4. Open source вклад

Поделиться полезным инструментом с сообществом.

Минимальная структура проекта

my_datalib/
├── setup.py                 # Конфигурация пакета
├── setup.cfg
├── pyproject.toml          # Современный подход (PEP 517)
├── requirements.txt         # Зависимости
├── README.md
├── LICENSE
├── my_datalib/             # Сам пакет
│   ├── __init__.py
│   ├── extractors.py
│   ├── transformers.py
│   └── loaders.py
└── tests/
    ├── test_extractors.py
    ├── test_transformers.py
    └── test_loaders.py

Шаг 1: Создать структуру проекта

mkdir my_datalib
cd my_datalib
git init

# Создать папку пакета
mkdir my_datalib
touch my_datalib/__init__.py

Шаг 2: Написать код модулей

# my_datalib/__init__.py
"""Data transformation library for common ETL operations."""

__version__ = "0.1.0"
__author__ = "Your Name"

from .extractors import extract_from_csv, extract_from_api
from .transformers import clean_data, normalize_columns
from .loaders import load_to_postgres, load_to_parquet

__all__ = [
    "extract_from_csv",
    "extract_from_api",
    "clean_data",
    "normalize_columns",
    "load_to_postgres",
    "load_to_parquet",
]

# my_datalib/extractors.py
import pandas as pd
import requests
from typing import Optional, Dict, Any

def extract_from_csv(filepath: str) -> pd.DataFrame:
    """Extract data from CSV file.
    
    Args:
        filepath: Path to CSV file
        
    Returns:
        Pandas DataFrame
        
    Raises:
        FileNotFoundError: If file doesn't exist
    """
    try:
        return pd.read_csv(filepath)
    except FileNotFoundError:
        raise FileNotFoundError(f"CSV file not found: {filepath}")

def extract_from_api(
    endpoint: str,
    auth_key: str,
    params: Optional[Dict[str, Any]] = None,
    timeout: int = 30
) -> Dict[str, Any]:
    """Extract data from REST API.
    
    Args:
        endpoint: API endpoint URL
        auth_key: API authentication key
        params: Query parameters (optional)
        timeout: Request timeout in seconds
        
    Returns:
        JSON response as dictionary
        
    Raises:
        requests.RequestException: If API call fails
    """
    headers = {"X-API-Key": auth_key, "Accept": "application/json"}
    
    try:
        response = requests.get(
            endpoint,
            headers=headers,
            params=params,
            timeout=timeout
        )
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        raise requests.RequestException(f"API error: {e}")

# my_datalib/transformers.py
import pandas as pd
import numpy as np
from typing import List

def clean_data(df: pd.DataFrame, drop_duplicates: bool = True) -> pd.DataFrame:
    """Clean dataframe by removing nulls and duplicates.
    
    Args:
        df: Input dataframe
        drop_duplicates: Whether to drop duplicate rows
        
    Returns:
        Cleaned dataframe
    """
    df = df.dropna()
    
    if drop_duplicates:
        df = df.drop_duplicates()
    
    return df.reset_index(drop=True)

def normalize_columns(
    df: pd.DataFrame,
    uppercase: bool = True,
    replace_spaces: bool = True
) -> pd.DataFrame:
    """Normalize column names.
    
    Args:
        df: Input dataframe
        uppercase: Convert to uppercase
        replace_spaces: Replace spaces with underscores
        
    Returns:
        Dataframe with normalized columns
    """
    columns = df.columns.tolist()
    
    if replace_spaces:
        columns = [col.replace(' ', '_') for col in columns]
    
    if uppercase:
        columns = [col.upper() for col in columns]
    
    df.columns = columns
    return df

# my_datalib/loaders.py
import pandas as pd
from sqlalchemy import create_engine, text
from typing import Optional
import logging

logger = logging.getLogger(__name__)

def load_to_postgres(
    df: pd.DataFrame,
    table_name: str,
    connection_string: str,
    if_exists: str = "append",
    index: bool = False
) -> None:
    """Load dataframe to PostgreSQL table.
    
    Args:
        df: Dataframe to load
        table_name: Target table name
        connection_string: PostgreSQL connection string
        if_exists: Action if table exists ('append', 'replace', 'fail')
        index: Whether to write index
    """
    engine = create_engine(connection_string)
    
    try:
        df.to_sql(
            table_name,
            con=engine,
            if_exists=if_exists,
            index=index,
            method="multi",  # Batch insert
            chunksize=1000    # Insert по 1000 строк
        )
        logger.info(f"Loaded {len(df)} rows to {table_name}")
    except Exception as e:
        logger.error(f"Failed to load to {table_name}: {e}")
        raise
    finally:
        engine.dispose()

def load_to_parquet(
    df: pd.DataFrame,
    filepath: str,
    compression: str = "snappy"
) -> None:
    """Load dataframe to Parquet file.
    
    Args:
        df: Dataframe to load
        filepath: Output file path
        compression: Compression algorithm (snappy, gzip, none)
    """
    try:
        df.to_parquet(
            filepath,
            compression=compression,
            index=False
        )
        logger.info(f"Saved {len(df)} rows to {filepath}")
    except Exception as e:
        logger.error(f"Failed to save to {filepath}: {e}")
        raise

Шаг 3: Написать тесты

# tests/test_extractors.py
import pytest
import pandas as pd
from my_datalib.extractors import extract_from_csv

def test_extract_from_csv_success(tmp_path):
    """Test successful CSV extraction."""
    # Arrange
    csv_file = tmp_path / "test.csv"
    csv_file.write_text("id,name\n1,Alice\n2,Bob")
    
    # Act
    df = extract_from_csv(str(csv_file))
    
    # Assert
    assert len(df) == 2
    assert list(df.columns) == ["id", "name"]
    assert df.loc[0, "name"] == "Alice"

def test_extract_from_csv_not_found():
    """Test error handling for missing file."""
    with pytest.raises(FileNotFoundError):
        extract_from_csv("/nonexistent/file.csv")

---

# tests/test_transformers.py
import pandas as pd
from my_datalib.transformers import clean_data, normalize_columns

def test_clean_data():
    """Test data cleaning."""
    df = pd.DataFrame({
        "id": [1, 2, 2, None],
        "name": ["Alice", "Bob", "Bob", "Charlie"]
    })
    
    result = clean_data(df, drop_duplicates=True)
    
    assert len(result) == 2  # Removed None and duplicate
    assert result.iloc[0]["id"] == 1.0

def test_normalize_columns():
    """Test column name normalization."""
    df = pd.DataFrame({"User ID": [1, 2], "First Name": ["a", "b"]})
    
    result = normalize_columns(df, uppercase=True, replace_spaces=True)
    
    assert list(result.columns) == ["USER_ID", "FIRST_NAME"]

Шаг 4: Конфигурация для упаковки

Modern approach с pyproject.toml:

# pyproject.toml
[build-system]
requires = ["setuptools>=45", "wheel", "setuptools_scm[toml]>=6.2"]
build-backend = "setuptools.build_meta"

[project]
name = "my-datalib"
version = "0.1.0"
description = "Common ETL utilities for data engineering"
readme = "README.md"
requires-python = ">=3.8"
authors = [{name = "Your Name", email = "you@example.com"}]
license = {text = "MIT"}
classifiers = [
    "Programming Language :: Python :: 3",
    "License :: OSI Approved :: MIT License",
    "Operating System :: OS Independent",
]

dependencies = [
    "pandas>=1.0",
    "requests>=2.25",
    "sqlalchemy>=1.4",
    "psycopg2-binary>=2.8",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.0",
    "pytest-cov>=3.0",
    "black",
    "ruff",
    "mypy",
]

[project.urls]
Homepage = "https://github.com/yourname/my-datalib"
Documentation = "https://my-datalib.readthedocs.io"
Repository = "https://github.com/yourname/my-datalib.git"

Шаг 5: Развертывание

Установка в режиме разработки:

pip install -e .[dev]  # Устанавливает с dependencies для разработки

Публикация на PyPI:

# Установить build tools
pip install build twine

# Создать distribution
python -m build

# Upload на PyPI (нужен аккаунт)
twine upload dist/*

Или в приватный PyPI:

twine upload --repository-url https://your-pypi.example.com dist/*

Шаг 6: Использование библиотеки

# В другом проекте
from my_datalib import extract_from_csv, clean_data, normalize_columns, load_to_postgres

# ETL pipeline
df = extract_from_csv("data.csv")
df = clean_data(df)
df = normalize_columns(df)
load_to_postgres(df, "target_table", "postgresql://user:pass@localhost/db")

Лучшие практики

1. Версионирование (Semantic Versioning):

• 0.1.0: major.minor.patch
• major: breaking changes
• minor: new features
• patch: bug fixes

2. Документация:

def my_function(param1: str, param2: int) -> bool:
    """Short description.
    
    Long description with details.
    
    Args:
        param1: Description
        param2: Description
        
    Returns:
        Description
        
    Raises:
        ValueError: When invalid
    """

3. Type hints:

def load_data(path: str, chunk_size: int = 1000) -> pd.DataFrame:
    ...

4. Логирование:

import logging

logger = logging.getLogger(__name__)
logger.info("Starting extraction")
logger.error(f"Error: {e}")

Заключение

Создание своей библиотеки:

• Экономит время и предотвращает дублирование
• Стандартизирует код в компании
• Улучшает качество через тесты и документацию
• Позволяет легко обновлять функциональность

Для Data Engineer'ов это критично для масштабирования ETL операций.