[4108] Сервис подписок: админка промокодов

alembic/versions/056_promocodes.py

@@ -0,0 +1,54 @@
+"""promocodes
+
+Revision ID: 056
+Revises: 055
+Create Date: 2026-01-14 13:05:54.325494
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "056"
+down_revision: Union[str, None] = "055"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "promocodes",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("title", sa.String(length=100), nullable=False),
+        sa.Column("code", sa.String(length=10), nullable=False),
+        sa.Column("valid_from", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("valid_until", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_promocodes")),
+        schema="xi_back_2",
+    )
+    op.create_index(
+        op.f("ix_xi_back_2_promocodes_code"),
+        "promocodes",
+        ["code"],
+        unique=True,
+        schema="xi_back_2",
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(
+        op.f("ix_xi_back_2_promocodes_code"),
+        table_name="promocodes",
+        schema="xi_back_2",
+    )
+    op.drop_table("promocodes", schema="xi_back_2")
+    # ### end Alembic commands ###

app/main.py

@@ -29,6 +29,7 @@
     posts,
     scheduler,
     storage_v2,
+    subscriptions,
     supbot,
     users,
 )
@@ -187,6 +188,7 @@ async def custom_swagger_ui_html() -> Response:
 app.include_router(supbot.api_router)
 app.include_router(classrooms.api_router)
 app.include_router(users.api_router)
+app.include_router(subscriptions.api_router)
 
 old_openapi = app.openapi
 

app/subscriptions/__init__.py

@@ -0,0 +1,3 @@
+from app.subscriptions.main import api_router
+
+__all__ = ["api_router"]

app/subscriptions/dependencies/promocodes_dep.py

@@ -0,0 +1,37 @@
+from typing import Annotated
+
+from fastapi import Depends, Path
+from starlette import status
+
+from app.common.fastapi_ext import Responses, with_responses
+from app.subscriptions.models.promocodes_db import Promocode
+
+
+class PromocodeResponses(Responses):
+    PROMOCODE_NOT_FOUND = status.HTTP_404_NOT_FOUND, "Promocode not found"
+
+
+@with_responses(PromocodeResponses)
+async def get_promocode_by_id(
+    promocode_id: Annotated[int, Path()],
+) -> Promocode:
+    promocode = await Promocode.find_first_by_id(promocode_id)
+    if promocode is None:
+        raise PromocodeResponses.PROMOCODE_NOT_FOUND
+    return promocode
+
+
+PromocodeByID = Annotated[Promocode, Depends(get_promocode_by_id)]
+
+
+@with_responses(PromocodeResponses)
+async def get_promocode_by_code(
+    code: Annotated[str, Path()],
+) -> Promocode:
+    promocode = await Promocode.find_first_by_kwargs(code=code)
+    if promocode is None:
+        raise PromocodeResponses.PROMOCODE_NOT_FOUND
+    return promocode
+
+
+PromocodeByCode = Annotated[Promocode, Depends(get_promocode_by_code)]

app/subscriptions/main.py

@@ -0,0 +1,39 @@
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from typing import Any
+
+from app.common.dependencies.api_key_dep import APIKeyProtection
+from app.common.dependencies.authorization_dep import ProxyAuthorized
+from app.common.dependencies.mub_dep import MUBProtection
+from app.common.fastapi_ext import APIRouterExt
+from app.subscriptions.routes import promocodes_mub
+
+outside_router = APIRouterExt(prefix="/api/public/subscription-service")
+
+authorized_router = APIRouterExt(
+    dependencies=[ProxyAuthorized],
+    prefix="/api/protected/subscription-service",
+)
+
+mub_router = APIRouterExt(
+    dependencies=[MUBProtection],
+    prefix="/mub/subscription-service",
+)
+mub_router.include_router(promocodes_mub.router)
+
+internal_router = APIRouterExt(
+    dependencies=[APIKeyProtection],
+    prefix="/internal/subscription-service",
+)
+
+
+@asynccontextmanager
+async def lifespan(_: Any) -> AsyncIterator[None]:
+    yield
+
+
+api_router = APIRouterExt(lifespan=lifespan)
+api_router.include_router(outside_router)
+api_router.include_router(authorized_router)
+api_router.include_router(mub_router)
+api_router.include_router(internal_router)

app/subscriptions/models/promocodes_db.py

@@ -0,0 +1,53 @@
+from datetime import datetime
+from typing import Annotated
+
+from pydantic import AwareDatetime, Field
+from pydantic_marshals.sqlalchemy import MappedModel
+from sqlalchemy import DateTime, String, select
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.common.config import Base
+from app.common.sqlalchemy_ext import db
+from app.common.utils.datetime import datetime_utc_now
+
+
+class Promocode(Base):
+    __tablename__ = "promocodes"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+
+    title: Mapped[str] = mapped_column(String(100))
+    code: Mapped[str] = mapped_column(String(10), index=True, unique=True)
+
+    valid_from: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), default=None
+    )
+    valid_until: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), default=None
+    )
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=datetime_utc_now
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=datetime_utc_now
+    )
+
+    TitleType = Annotated[str, Field(min_length=1, max_length=100)]
+    CodeType = Annotated[str, Field(min_length=1, max_length=10)]
+
+    InputSchema = MappedModel.create(
+        columns=[
+            (title, TitleType),
+            (code, CodeType),
+            (valid_from, AwareDatetime | None),
+            (valid_until, AwareDatetime | None),
+        ]
+    )
+    ResponseSchema = InputSchema.extend(
+        columns=[id, (created_at, AwareDatetime), (updated_at, AwareDatetime)]
+    )
+
+    @classmethod
+    async def is_present_by_code(cls, code: str) -> bool:
+        return await db.is_present(select(cls).filter_by(code=code))

app/subscriptions/routes/promocodes_mub.py

@@ -0,0 +1,99 @@
+from collections.abc import Sequence
+from typing import Annotated, Self
+
+from fastapi import Query
+from pydantic import model_validator
+from starlette import status
+
+from app.common.fastapi_ext import APIRouterExt, Responses
+from app.common.utils.datetime import datetime_utc_now
+from app.subscriptions.dependencies.promocodes_dep import PromocodeByCode, PromocodeByID
+from app.subscriptions.models.promocodes_db import Promocode
+
+router = APIRouterExt(tags=["promocodes mub"])
+
+
+@router.get(
+    "/promocodes/",
+    response_model=list[Promocode.ResponseSchema],
+    summary="List paginated promocodes",
+)
+async def list_promocodes(
+    offset: Annotated[int, Query(ge=0)] = 0,
+    limit: Annotated[int, Query(ge=1, le=100)] = 100,
+) -> Sequence[Promocode]:
+    return await Promocode.find_paginated_by_kwargs(
+        offset, limit, Promocode.created_at.desc()
+    )
+
+
+class PromocodeInputSchema(Promocode.InputSchema):
+    @model_validator(mode="after")
+    def validate_promocode_valid_from_and_until_date(self) -> Self:
+        if (
+            self.valid_from is not None and self.valid_until is not None
+        ) and self.valid_from >= self.valid_until:
+            raise ValueError("the end date cannot be earlier than the start date")
+        return self
+
+
+class PromocodeConflictResponses(Responses):
+    PROMOCODE_ALREADY_EXISTS = status.HTTP_409_CONFLICT, "Promocode already exists"
+
+
+@router.post(
+    "/promocodes/",
+    status_code=status.HTTP_201_CREATED,
+    response_model=Promocode.ResponseSchema,
+    responses=PromocodeConflictResponses.responses(),
+    summary="Create a new promocode",
+)
+async def create_promocode(data: PromocodeInputSchema) -> Promocode:
+    if await Promocode.is_present_by_code(code=data.code):
+        raise PromocodeConflictResponses.PROMOCODE_ALREADY_EXISTS
+    return await Promocode.create(**data.model_dump())
+
+
+@router.get(
+    "/promocodes/by-id/{promocode_id}/",
+    response_model=Promocode.ResponseSchema,
+    summary="Retrieve any promocode by id",
+)
+async def retrieve_promocode_by_id(promocode: PromocodeByID) -> Promocode:
+    return promocode
+
+
+@router.get(
+    "/promocodes/by-code/{code}/",
+    response_model=Promocode.ResponseSchema,
+    summary="Retrieve any promocode by code",
+)
+async def retrieve_promocode_by_code(promocode: PromocodeByCode) -> Promocode:
+    return promocode
+
+
+@router.put(
+    "/promocodes/{promocode_id}/",
+    response_model=Promocode.ResponseSchema,
+    responses=PromocodeConflictResponses.responses(),
+    summary="Update any promocode by id",
+)
+async def put_promocode(
+    promocode: PromocodeByID,
+    data: PromocodeInputSchema,
+) -> Promocode:
+    if data.code != promocode.code and await Promocode.is_present_by_code(
+        code=data.code
+    ):
+        raise PromocodeConflictResponses.PROMOCODE_ALREADY_EXISTS
+    promocode.update(**data.model_dump(), updated_at=datetime_utc_now())
+    return promocode
+
+
+@router.delete(
+    "/promocodes/{promocode_id}/",
+    status_code=status.HTTP_204_NO_CONTENT,
+    summary="Delete any promocode by id",
+)
+async def delete_promocode(promocode: PromocodeByID) -> None:
+    await promocode.delete()

pyproject.toml

@@ -78,7 +78,7 @@ dlint = "0.14.0"
 [tool.poetry.group.tests.dependencies]
 pytest = "^8.3.5"
 pytest-cov = "^6.1.1"
-pytest-lazy-fixtures = "^1.3.4"
+pytest-lazy-fixtures = "^1.4.0"
 pydantic-marshals = {extras = ["assert-contains"], version = "0.3.18"}
 freezegun = "^1.5.1"
 respx = "^0.22.0"

tests/subscriptions/conftest.py

@@ -0,0 +1,41 @@
+import pytest
+from faker import Faker
+
+from app.subscriptions.models.promocodes_db import Promocode
+from tests.common.active_session import ActiveSession
+from tests.common.types import AnyJSON
+from tests.subscriptions import factories
+
+
+@pytest.fixture()
+async def promocode(active_session: ActiveSession, faker: Faker) -> Promocode:
+    async with active_session():
+        return await Promocode.create(
+            **factories.LimitedPromocodeInputFactory.build_python(
+                code=faker.pystr(min_chars=10, max_chars=10),
+            )
+        )
+
+
+@pytest.fixture()
+async def promocode_data(promocode: Promocode) -> AnyJSON:
+    return Promocode.ResponseSchema.model_validate(promocode).model_dump(mode="json")
+
+
+@pytest.fixture()
+async def other_promocode(active_session: ActiveSession, faker: Faker) -> Promocode:
+    async with active_session():
+        return await Promocode.create(
+            **factories.LimitedPromocodeInputFactory.build_python(
+                code=faker.pystr(min_chars=9, max_chars=9),
+            )
+        )
+
+
+@pytest.fixture()
+async def deleted_promocode(
+    active_session: ActiveSession, promocode: Promocode
+) -> Promocode:
+    async with active_session():
+        await promocode.delete()
+    return promocode