🛠️ Технологический стек¶

# Пример FastAPI эндпоинта
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from typing import List

app = FastAPI(title="Safety API", version="1.0.0")

class DocumentCreate(BaseModel):
    name: str
    description: str | None = None
    category: str

@app.post("/documents/", response_model=DocumentResponse)
async def create_document(
    document: DocumentCreate,
    current_user: User = Depends(get_current_active_user),
    db: Session = Depends(get_db)
):
    """Создание нового документа"""
    db_document = Document(**document.dict(), uploaded_by=current_user.id)
    db.add(db_document)
    db.commit()
    db.refresh(db_document)
    return db_document

# Пример модели SQLAlchemy
from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "safety_users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True, nullable=False)
    email = Column(String(100), unique=True, index=True, nullable=False)
    full_name = Column(String(200), nullable=False)
    password_hash = Column(String(255), nullable=False)
    role = Column(String(20), default="user")
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

# Пример аутентификации
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# Пример обработки Excel файла
import openpyxl
from openpyxl import load_workbook

def extract_excel_fields(file_path: str) -> dict:
    """Извлечение полей из Excel шаблона"""
    workbook = load_workbook(file_path)
    fields = {}

    for sheet_name in workbook.sheetnames:
        sheet = workbook[sheet_name]
        sheet_fields = []

        for row in sheet.iter_rows():
            for cell in row:
                if cell.value and isinstance(cell.value, str):
                    if '{' in cell.value and '}' in cell.value:
                        # Извлекаем поля в фигурных скобках
                        import re
                        matches = re.findall(r'\{([^}]+)\}', cell.value)
                        sheet_fields.extend(matches)

        fields[sheet_name] = list(set(sheet_fields))

    return fields

# Пример интеграции с FNS API
import httpx
from typing import Optional

class FNSService:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api-fns.ru/api"

    async def get_organization_data(self, inn: str) -> Optional[dict]:
        """Получение данных организации из ФНС"""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/egr",
                params={"req": inn, "key": self.api_key}
            )
            if response.status_code == 200:
                return response.json()
        return None

# Пример отправки email
import aiosmtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

async def send_email(to: str, subject: str, body: str):
    """Отправка email уведомления"""
    message = MIMEMultipart()
    message["From"] = SMTP_USER
    message["To"] = to
    message["Subject"] = subject
    message.attach(MIMEText(body, "html"))

    await aiosmtplib.send(
        message,
        hostname=SMTP_HOST,
        port=SMTP_PORT,
        username=SMTP_USER,
        password=SMTP_PASSWORD,
        use_tls=True
    )

# Пример настройки логирования
import logging
import structlog
from pythonjsonlogger import jsonlogger

# Настройка структурированного логирования
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

# Пример теста
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.database import get_db
from tests.factories import UserFactory

client = TestClient(app)

@pytest.fixture
def test_user():
    return UserFactory()

def test_create_document(test_user):
    response = client.post(
        "/documents/",
        json={"name": "Test Document", "category": "safety"},
        headers={"Authorization": f"Bearer {test_user.token}"}
    )
    assert response.status_code == 201
    assert response.json()["name"] == "Test Document"