Какие используешь классы в FastAPI?

Question

claude-haiku-4.5 · Accepted Answer

## Классы в FastAPI

FastAPI активно использует различные классы для построения надёжных и типобезопасных API приложений. Вот основные категории классов, которые я использую в повседневной разработке.

### 1. Pydantic модели (BaseModel)

Основной способ определения схем данных и валидации:

```python
from pydantic import BaseModel, Field, validator
from typing import Optional, List

class UserCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: Optional[int] = Field(None, ge=0, le=150)
    
    @validator('name')
    def name_not_empty(cls, v):
        if not v.strip():
            raise ValueError('Имя не может быть пустым')
        return v.title()

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    age: Optional[int]
    
    class Config:
        from_attributes = True  # ORM mode для SQLAlchemy
```

### 2. APIRouter

Для организации маршрутов в отдельные файлы:

```python
from fastapi import APIRouter, Depends, HTTPException
from typing import List

router = APIRouter(
    prefix='/api/v1/users',
    tags=['users'],
    responses={404: {'description': 'Not found'}}
)

@router.get('', response_model=List[UserResponse])
async def get_users(skip: int = 0, limit: int = 10):
    """Получить список всех пользователей."""
    users = db.query(User).offset(skip).limit(limit).all()
    return users

@router.post('', response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    """Создать нового пользователя."""
    db_user = User(**user.dict())
    db.add(db_user)
    db.commit()
    return db_user
```

### 3. Dependency Injection (Depends)

Для внедрения зависимостей:

```python
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session

def get_db():
    """Зависимость для получения сессии БД."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> User:
    """Получить текущего аутентифицированного пользователя."""
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    username: str = payload.get('sub')
    if username is None:
        raise HTTPException(status_code=401, detail='Invalid credentials')
    
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise HTTPException(status_code=404, detail='User not found')
    
    return user

@router.get('/me', response_model=UserResponse)
async def get_profile(current_user: User = Depends(get_current_user)):
    """Получить профиль текущего пользователя."""
    return current_user
```

### 4. HTTPException

Для возврата HTTP ошибок:

```python
from fastapi import HTTPException
from starlette import status

@router.get('/{user_id}', response_model=UserResponse)
async def get_user(user_id: int, db: Session = Depends(get_db)):
    """Получить пользователя по ID."""
    user = db.query(User).filter(User.id == user_id).first()
    
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail='User not found'
        )
    
    return user
```

### 5. Request и Response

Для работы с запросами и ответами:

```python
from fastapi import Request, Response
from fastapi.responses import JSONResponse, FileResponse, StreamingResponse

@router.post('/upload')
async def upload_file(request: Request):
    """Обработать загруженный файл."""
    body = await request.body()
    return {'size': len(body)}

@router.get('/download')
async def download_file():
    """Скачать файл."""
    return FileResponse(
        path='data.csv',
        filename='exported_data.csv',
        media_type='text/csv'
    )

@router.get('/stream')
async def stream_data():
    """Потоковая передача данных."""
    async def generate():
        for i in range(10):
            yield f'data: {i}
'
    
    return StreamingResponse(generate(), media_type='text/event-stream')
```

### 6. UploadFile

Для загрузки файлов:

```python
from fastapi import UploadFile, File

@router.post('/upload-file')
async def upload_single_file(file: UploadFile = File(...)):
    """Загрузить один файл."""
    contents = await file.read()
    return {
        'filename': file.filename,
        'content_type': file.content_type,
        'size': len(contents)
    }

@router.post('/upload-multiple')
async def upload_multiple_files(files: List[UploadFile] = File(...)):
    """Загрузить несколько файлов."""
    results = []
    for file in files:
        contents = await file.read()
        results.append({
            'filename': file.filename,
            'size': len(contents)
        })
    return results
```

### 7. Query параметры

```python
from fastapi import Query

@router.get('/search')
async def search_users(
    q: str = Query(..., min_length=1, max_length=50),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    sort: str = Query('name', regex='^(name|email|age)$')
):
    """Поиск пользователей с фильтрацией."""
    return {'query': q, 'skip': skip, 'limit': limit, 'sort': sort}
```

### 8. OAuth2PasswordBearer

Для аутентификации с токенами:

```python
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from passlib.context import CryptContext
import jwt
from datetime import datetime, timedelta

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')
pwd_context = CryptContext(schemes=['bcrypt'])

SECRET_KEY = 'your-secret-key-change-me'
ALGORITHM = 'HS256'

@router.post('/token')
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    """Получить JWT токен."""
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail='Invalid credentials')
    
    access_token_expires = timedelta(hours=24)
    expire = datetime.utcnow() + access_token_expires
    token_data = {'sub': user.username, 'exp': expire}
    encoded_jwt = jwt.encode(token_data, SECRET_KEY, algorithm=ALGORITHM)
    
    return {'access_token': encoded_jwt, 'token_type': 'bearer'}
```

### 9. Background Tasks

Для выполнения фоновых задач:

```python
from fastapi import BackgroundTasks
import time

def send_email_async(email: str, message: str):
    """Отправить email в фоне (имитация)."""
    time.sleep(2)
    print(f'Email sent to {email}: {message}')

@router.post('/send-notification')
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks
):
    """Отправить уведомление и вернуться немедленно."""
    background_tasks.add_task(send_email_async, email, 'Hello!')
    return {'message': 'Notification scheduled'}
```

### 10. SQLAlchemy модели

Основные классы для работы с БД:

```python
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    password_hash = Column(String(255), nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    def __repr__(self):
        return f'<User(id={self.id}, name={self.name}, email={self.email})>'
```

### 11. Класс FastAPI

Главное приложение:

```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(
    title='My API',
    description='API documentation',
    version='1.0.0'
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=['http://localhost:3000'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*']
)

app.include_router(router, prefix='/api/v1')

@app.get('/')
async def root():
    return {'message': 'Hello World'}
```

### 12. Кастомные исключения

```python
from fastapi import HTTPException
from starlette import status

class UserNotFound(HTTPException):
    def __init__(self, user_id: int):
        super().__init__(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f'User with id {user_id} not found'
        )

class InsufficientPermissions(HTTPException):
    def __init__(self):
        super().__init__(
            status_code=status.HTTP_403_FORBIDDEN,
            detail='Insufficient permissions'
        )

@router.delete('/{user_id}')
async def delete_user(
    user_id: int,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """Удалить пользователя."""
    if current_user.id != user_id and not current_user.is_admin:
        raise InsufficientPermissions()
    
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise UserNotFound(user_id)
    
    db.delete(user)
    db.commit()
    return {'detail': 'User deleted'}
```

Все эти классы вместе создают мощную и гибкую систему для разработки профессиональных API приложений на Python.

Какие используешь классы в FastAPI?

Комментарии (1)

Классы в FastAPI

1. Pydantic модели (BaseModel)

2. APIRouter

3. Dependency Injection (Depends)

4. HTTPException

5. Request и Response

6. UploadFile

7. Query параметры

8. OAuth2PasswordBearer

9. Background Tasks

10. SQLAlchemy модели

11. Класс FastAPI

12. Кастомные исключения