Лабораторная работа 1

Цель работы

Реализовать полноценное серверное приложение с помощью фреймворка FastAPI с применением дополнительных средств и библиотек: SQLAlchemy async ORM, PostgreSQL, Alembic, JWT-авторизация.

Тема варианта — сервис управления личными финансами: учёт доходов и расходов, категории, бюджеты, цели, теги.

Выполненные практики

Лабораторная работа выполнена целостно: функциональность практик 1.1, 1.2 и 1.3 реализована в рамках одного приложения и одного итогового этапа разработки.

Стек технологий

Структура проекта

Модель данных

Реализовано 6 таблиц с отношениями one-to-many и many-to-many.

Подключение к БД

# app/core/database.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from .config import DATABASE_URL, DB_ECHO

engine = create_async_engine(DATABASE_URL, echo=DB_ECHO)

async_session_maker = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_session() -> AsyncSession:
    async with async_session_maker() as session:
        yield session

Пример модели с отношениями

# app/models/models.py
class Transaction(Base):
    __tablename__ = "transaction"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    amount: Mapped[float] = mapped_column(Float, nullable=False)
    description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    date: Mapped[date] = mapped_column(Date, nullable=False)

    user_id: Mapped[int] = mapped_column(ForeignKey("user.id", ondelete="CASCADE"))
    category_id: Mapped[int] = mapped_column(ForeignKey("category.id"))

    user: Mapped["User"] = relationship("User", back_populates="transactions", lazy="selectin")
    category: Mapped["Category"] = relationship("Category", lazy="selectin")
    tags: Mapped[List["Tag"]] = relationship(
        "Tag", secondary="transaction_tag", lazy="selectin"
    )

Generic BaseCrud

Все CRUD-классы наследуются от общего BaseCrud, параметризованного через Generic. Специфичная логика добавляется в подклассах.

# app/crud/base.py
class BaseCrud(Generic[ModelT, GetSchemaT, CreateSchemaT, UpdateSchemaT]):
    base_model: ClassVar[Type[ModelT]]
    get_schema: ClassVar[Type[GetSchemaT]]

    @classmethod
    async def get_by_id(cls, session, record_id) -> Optional[GetSchemaT]: ...

    @classmethod
    async def get_all(cls, session, offset=0, limit=100) -> List[GetSchemaT]: ...

    @classmethod
    async def create(cls, session, **kwargs) -> GetSchemaT: ...

    @classmethod
    async def update(cls, session, record_id, **kwargs) -> Optional[GetSchemaT]: ...

    @classmethod
    async def delete(cls, session, record_id) -> None: ...

# app/crud/user.py — пример наследования
class UserCrud(BaseCrud[User, UserRead, UserCreate, UserUpdate]):
    base_model = User
    get_schema = UserRead

    @classmethod
    async def get_by_username(cls, session, username) -> Optional[UserRead]: ...

    @classmethod
    async def create_user(cls, session, data: UserCreate) -> UserRead: ...

Авторизация

Авторизация реализована вручную без сторонних auth-библиотек. Используются PyJWT для токенов и passlib[bcrypt] для паролей.

Хэширование паролей

# app/core/security.py
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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

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

Генерация и верификация JWT

def create_access_token(payload: dict) -> str:
    data = payload.copy()
    data["exp"] = datetime.now(timezone.utc) + timedelta(minutes=JWT_EXPIRE_MINUTES)
    return jwt.encode(data, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)

def decode_access_token(token: str) -> dict:
    return jwt.decode(token, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM])

Dependency получения текущего пользователя

# app/api/dependencies.py
async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer()),
    session: AsyncSession = Depends(get_session),
) -> UserRead:
    token = credentials.credentials
    payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM])
    user_id = payload.get("sub")
    user = await UserCrud.get_by_id(session, user_id)
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return user

API Эндпоинты

Auth

Users

Categories

Tags

Transactions

При невалидных внешних ключах, конфликте уникальности или передаче несуществующих tag_ids API возвращает 400 Bad Request.

Budgets

Goals

Запуск проекта

# 1. Запустить PostgreSQL
docker-compose up -d

# 2. Установить зависимости
pip install -r requirements.txt

# 3. Применить миграции
alembic upgrade head

# 4. Запустить сервер
uvicorn app.main:app --reload

После запуска документация доступна по адресу http://localhost:8000/docs