backend/src/service/node_packages.py - master - nollieundergrob/PythonPackageStorage

backend/src/service/node_packages.py
122 строки4 KB
nollieundergrob
GENERAL FIX BACKEND
15 авг 2025, 16:01
15 авг 2025, 16:01a8b7dd5
# import fake.user as data
# from routers.user import User
from schemas.nodejs import Library as Schema
from schemas.nodejs import LibraryCreate, LibraryModify, DetailLibrary
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update, or_
from data.models.nodejs import NodeJSLibrary as Library
from sqlalchemy.engine import Result
from fastapi.exceptions import HTTPException
from fastapi import status
import asyncio
from utils.parse.nodejs_version import create_and_delete
import os
from config.settings import settings
# Update the code with the new data
class LibraryService:
    """READ"""

    @staticmethod
    async def get_all(session: AsyncSession) -> list[Schema]:
        stmt = select(Library).order_by(Library.id)
        result: Result = await session.execute(stmt)
        libs = result.scalars().all()
        return [Schema.model_validate(user) for user in list(libs)]

    @staticmethod
    async def get_detail(library_id: int, session: AsyncSession) -> Schema:
        try:
            library = await session.get(Library, library_id)
            if library is None:
                raise ValueError("Not Found")

            try:
                # Получаем список версий и преобразуем в список словарей
                a = settings.BASE_PATH / f"/storage/npm/{library.name}"
                version_files = os.listdir(settings.BASE_PATH / f"storage/npm/{library.name}")
                versions = [file for file in version_files]  # Формируем словари
            except FileNotFoundError as e:
                versions = []

            return DetailLibrary(library=library, version=versions)
        except ValueError as e:
            raise HTTPException(detail=str(e), status_code=status.HTTP_404_NOT_FOUND)

    """CREATE"""

    @staticmethod
    async def create(session: AsyncSession, library_in: LibraryCreate) -> Library:
        library = Library(**library_in.model_dump())
        exists_lib = await session.execute(select(Library).where(Library.name == library_in.name))
        if exists_lib.scalar_one_or_none():
            raise ValueError("Library already exists")

        session.add(library)
        await create_and_delete(library.name)

        await session.commit()
        await session.refresh(library)
        return library


    @staticmethod
    async def modify(
        session: AsyncSession, library_id: int, library_in: LibraryModify
    ) -> Library:
        existing_user = await session.execute(
            select(Library).where(Library.id == library_id)
        )
        if not existing_user.scalar():
            raise ValueError("Not Found")

        library_data = library_in.model_dump(exclude_unset=True)

        if library_data:
            await session.execute(
                update(Library).where(Library.id == library_id).values(**library_data)
            )
            await session.commit()

        updated_user = await session.execute(
            select(Library).where(Library.id == library_id)
        )
        return updated_user.scalar_one()

    @staticmethod
    async def replace(
        session: AsyncSession, library_id: int, library_in: LibraryCreate
    ) -> Library:
        library_data = library_in.model_dump(exclude_unset=True)
        await session.execute(
            update(Library).where(Library.id == library_id).values(**library_data)
        )
        await session.commit()
        updated_user = await session.execute(
            select(Library).where(Library.id == library_id)
        )
        return updated_user.scalar_one()

    @staticmethod
    async def delete(
        session: AsyncSession,
        library_id: int,
        # current_user: Library
    ) -> Library:
        # Получаем пользователя для удаления
        result = await session.execute(select(Library).where(Library.id == library_id))
        user_to_delete = result.scalar_one_or_none()

        if not user_to_delete:
            raise ValueError("User not found")

        # if current_user.id != user_to_delete.id and not current_user.is_admin:
        # raise PermissionError("Forbiden")

        # Удаляем пользователя
        await session.delete(user_to_delete)
        await session.commit()

        return user_to_delete


service = LibraryService()