webapp/app/services/remote_server_service.py

from typing import List, Optional
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models.remote_server import RemoteServer
from app.schemas.remote_server import RemoteServerCreate, RemoteServerUpdate


class RemoteServerService:
    @staticmethod
    async def get_all(db: AsyncSession) -> List[RemoteServer]:
        result = await db.execute(select(RemoteServer))
        return result.scalars().all()

    @staticmethod
    async def get_by_server_id(db: AsyncSession, server_id: str) -> Optional[RemoteServer]:
        result = await db.execute(select(RemoteServer).filter(RemoteServer.server_id == server_id))
        return result.scalar_one_or_none()

    @staticmethod
    async def create(db: AsyncSession, server_in: RemoteServerCreate) -> RemoteServer:
        obj = RemoteServer(**server_in.model_dump())
        db.add(obj)
        await db.commit()
        await db.refresh(obj)
        return obj

    @staticmethod
    async def update(db: AsyncSession, server_id: str, server_in: RemoteServerUpdate) -> Optional[RemoteServer]:
        obj = await RemoteServerService.get_by_server_id(db, server_id)
        if not obj:
            return None
        
        update_data = server_in.model_dump(exclude_unset=True)
        for field, value in update_data.items():
            setattr(obj, field, value)
        
        db.add(obj)
        await db.commit()
        await db.refresh(obj)
        return obj

    @staticmethod
    async def delete(db: AsyncSession, server_id: str) -> bool:
        obj = await RemoteServerService.get_by_server_id(db, server_id)
        if not obj:
            return False
        await db.delete(obj)
        await db.commit()
        return True