src/app/api/db_cmd.py - master - Ulyanoff/mlops2

src/app/api/db_cmd.py
61 строка3 KB
Artem Ivochkin
Initial commit
28 ноя 2025, 21:56
28 ноя 2025, 21:56509a1cb
# -*- coding: utf-8 -*-
import logging
from typing import Literal

from fastapi import Request, Depends, Path
from fastapi.responses import JSONResponse
from sqlalchemy.ext.asyncio import AsyncSession, AsyncConnection

from app.api.routes import v0
from database.core import get_session, get_conn
from database.tables import DBase, Users
from serv.consts import common, default_users
from serv.log_utils import trunc_str

logger = logging.getLogger(__name__)


@v0.post("/db/{cmd}",
         summary="Создание и наполнение базы",
         description="Создаёт или удаляет таблицы, добавляет тестовых юзеров",
         responses=common,
         )
async def db_cmd(req: Request,
                 cmd: str = Path(...,
                                 description="Команды для базы данных",
                                 example='create - создание пустых таблиц, для наполнения: demo'),
                 # cmd: Literal['drop', 'demo', 'create'] = Path(..., description="Команды для базы данных",
                 #                                                example='create - создание пустых таблиц, '
                 #                                                        'для наполнения: demo'),
                 sess: AsyncSession = Depends(get_session),
                 conn: AsyncConnection = Depends(get_conn),
                 ):
    """Создание и наполнение тестовой базы"""
    api_path = req.url.path  # путь вызова API
    logger.debug(f"\n\t=>\t\t{api_path} ({db_cmd.__doc__}) started...")

    match cmd:
        case 'drop':
            await conn.run_sync(DBase.metadata.drop_all)
            await conn.commit()
            logger.fatal('All DB-tables dropped')
            stat, res = 200, {"dropped": "All DB-tables dropped"}
        case 'create':
            await conn.run_sync(DBase.metadata.create_all)
            await conn.commit()
            logger.info('DB-tables created')
            stat, res = 200, {"created": "DB-tables created"}
        case 'demo':
            user_list = list()
            for user in default_users:
                user_list.append(Users(name=f"auto {user['name']}", email=f"{user['email']}", psw=f"{user['psw']}"))
            sess.add_all(user_list)
            await sess.commit()
            logger.info('Demo data uploaded')
            stat, res = 200, {"uploaded": "Demo data uploaded"}
        case _:
            stat, res = 409, {"error": f"Unexpected command: {cmd}"}

    ll = f"\n\t<=\t\t{api_path}, HTTP={stat} ended\n\tresp\t{trunc_str(res)}"
    logger.debug(ll) if 199 < stat < 300 else logger.error(ll)
    return JSONResponse(res, status_code=stat)