REST API для мобильного приложения: FastAPI + PostgreSQL + JWT

pip install fastapi uvicorn sqlalchemy psycopg2-binary pydantic jwt python-jose passlib

api/
├── main.py
├── config.py
├── database.py
├── models.py
├── schemas.py
├── auth.py
├── dependencies.py
└── routers/
    ├── users.py
    └── items.py

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://user:pass@localhost:5432/mobile_app"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# models.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    username = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    description = Column(String, nullable=True)
    owner_id = Column(Integer, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

# schemas.py
from pydantic import BaseModel, EmailStr
from datetime import datetime

class UserCreate(BaseModel):
    email: EmailStr
    username: str
    password: str

class UserResponse(BaseModel):
    id: int
    email: str
    username: str
    is_active: bool
    created_at: datetime

    class Config:
        from_attributes = True

class ItemCreate(BaseModel):
    title: str
    description: str = None

class ItemResponse(BaseModel):
    id: int
    title: str
    description: str
    owner_id: int
    created_at: datetime

    class Config:
        from_attributes = True

# auth.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE = 3600  # 1 час

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

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(seconds=ACCESS_TOKEN_EXPIRE)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_token(token: str) -> dict:
    return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

# dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from database import get_db
from models import User
from auth import decode_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> User:
    try:
        payload = decode_token(token)
        email: str = payload.get("sub")
        if email is None:
            raise HTTPException(status_code=401, detail="Невалидный токен")
    except JWTError:
        raise HTTPException(status_code=401, detail="Невалидный токен")

    user = db.query(User).filter(User.email == email).first()
    if user is None or not user.is_active:
        raise HTTPException(status_code=401, detail="Пользователь не найден")
    return user

# routers/auth.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db, SessionLocal
from models import User
from auth import hash_password, verify_password, create_access_token

router = APIRouter(prefix="/auth", tags=["auth"])

@router.post("/register")
def register(username: str, email: str, password: str, db: Session = Depends(get_db)):
    # проверка дубликатов
    existing = db.query(User).filter(
        (User.email == email) | (User.username == username)
    ).first()
    if existing:
        raise HTTPException(status_code=400, detail="Email или username заняты")

    user = User(
        email=email,
        username=username,
        hashed_password=hash_password(password)
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return {"id": user.id, "email": user.email, "username": user.username}

@router.post("/login")
def login(email: str, password: str, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.email == email).first()
    if not user or not verify_password(password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Неверные данные")

    token = create_access_token(data={"sub": user.email})
    return {"access_token": token, "token_type": "bearer"}

# routers/items.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from database import get_db
from models import User, Item
from schemas import ItemCreate, ItemResponse
from dependencies import get_current_user

router = APIRouter(prefix="/items", tags=["items"])

@router.get("/", response_model=list(ItemResponse))
def list_items(db: Session = Depends(get_db), user: User = Depends(get_current_user)):
    return db.query(Item).filter(Item.owner_id == user.id).all()

@router.post("/", response_model=ItemResponse)
def create_item(
    item: ItemCreate,
    db: Session = Depends(get_db),
    user: User = Depends(get_current_user)
):
    db_item = Item(**item.dict(), owner_id=user.id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@router.delete("/{item_id}")
def delete_item(
    item_id: int,
    db: Session = Depends(get_db),
    user: User = Depends(get_current_user)
):
    item = db.query(Item).filter(
        Item.id == item_id, Item.owner_id == user.id
    ).first()
    if not item:
        raise HTTPException(status_code=404, detail="Не найдено")
    db.delete(item)
    db.commit()
    return {"detail": "Удалено"}

# main.py
from fastapi import FastAPI
from database import Base, engine
from routers import auth, items

Base.metadata.create_all(bind=engine)

app = FastAPI(title="Mobile App API", version="1.0.0")

app.include_router(auth.router)
app.include_router(items.router)

@app.get("/health")
def health():
    return {"status": "ok"}

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

import requests

# Регистрация
resp = requests.post("http://localhost:8000/auth/register", json={
    "email": "test@example.com",
    "username": "testuser",
    "password": "secret123"
})
print(resp.json())

# Получение токена
resp = requests.post("http://localhost:8000/auth/login", json={
    "email": "test@example.com",
    "password": "secret123"
})
token = resp.json()["access_token"]

# Создание объекта (защищённый эндпоинт)
headers = {"Authorization": f"Bearer {token}"}
resp = requests.post("http://localhost:8000/items/", json={
    "title": "Мой первый объект",
    "description": "Создано с мобильного"
}, headers=headers)
print(resp.json())