Init комит шаблона FastAPI

fastapi-app/.dockeringore

@@ -0,0 +1,3 @@
+venv
+Dockerfile
+tasks.db

fastapi-app/.gitignore

@@ -0,0 +1,4 @@
+venv
+tasks.db
+.idea
+__pycache__

fastapi-app/Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.11-slim
+
+COPY ../../../../../AppData/Local/Temp .
+
+RUN pip install -r requirements.txt
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]
+
+# Для запуска введите две команды:
+# docker build . --tag fastapi_app
+# docker run -p 80:80 fastapi_app
+
+# Или одной командой
+# docker build . --tag fastapi_app && docker run -p 80:80 fastapi_app

fastapi-app/database.py

@@ -0,0 +1,30 @@
+from typing import Optional
+
+from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+engine = create_async_engine("sqlite+aiosqlite:///tasks.db")
+new_session = async_sessionmaker(engine, expire_on_commit=False)
+
+
+class Model(DeclarativeBase):
+    pass
+
+
+class TaskOrm(Model):
+    __tablename__ = "tasks"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str]
+    description: Mapped[Optional[str]]
+
+
+async def create_tables():
+    # https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html#synopsis-core
+    async with engine.begin() as conn:
+        await conn.run_sync(Model.metadata.create_all)
+
+
+async def delete_tables():
+    async with engine.begin() as conn:
+        await conn.run_sync(Model.metadata.drop_all)

fastapi-app/main.py

@@ -0,0 +1,20 @@
+from fastapi import FastAPI
+
+from contextlib import asynccontextmanager
+
+from database import create_tables, delete_tables
+from router import router as tasks_router
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    await delete_tables()
+    print("База очищена")
+    await create_tables()
+    print("База готова к работе")
+    yield
+    print("Выключение")
+
+
+app = FastAPI(lifespan=lifespan)
+app.include_router(tasks_router)

fastapi-app/repository.py

@@ -0,0 +1,26 @@
+from sqlalchemy import select
+
+from database import new_session, TaskOrm
+from schemas import STaskAdd, STask
+
+
+class TaskRepository:
+    @classmethod
+    async def add_one(cls, data: STaskAdd) -> int:
+        async with new_session() as session:
+            task_dict = data.model_dump()
+
+            task = TaskOrm(**task_dict)
+            session.add(task)
+            await session.flush()
+            await session.commit()
+            return task.id
+
+    @classmethod
+    async def find_all(cls) -> list[STask]:
+        async with new_session() as session:
+            query = select(TaskOrm)
+            result = await session.execute(query)
+            task_models = result.scalars().all()
+            task_schemas = [STask.model_validate(task_model) for task_model in task_models]
+            return task_schemas

fastapi-app/router.py

@@ -0,0 +1,25 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends
+
+from repository import TaskRepository
+from schemas import STaskAdd, STask, STaskId
+
+router = APIRouter(
+    prefix="/tasks",
+    tags=["Таски"],
+)
+
+
+@router.post("")
+async def add_task(
+        task: Annotated[STaskAdd, Depends()],
+) -> STaskId:
+    task_id = await TaskRepository.add_one(task)
+    return {"ok": True, "task_id": task_id}
+
+
+@router.get("")
+async def get_tasks() -> list[STask]:
+    tasks = await TaskRepository.find_all()
+    return tasks

fastapi-app/schemas.py

@@ -0,0 +1,19 @@
+from typing import Optional
+
+from pydantic import BaseModel, ConfigDict
+
+
+class STaskAdd(BaseModel):
+    name: str
+    description: Optional[str] = None
+
+
+class STask(STaskAdd):
+    id: int
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class STaskId(BaseModel):
+    ok: bool = True
+    task_id: int