SupplyAgent/supply_infra/db/repositories/oss_log_repo.py

from __future__ import annotations

from sqlalchemy import distinct, select

from supply_infra.db.models.oss_log import OssLog
from supply_infra.db.repositories.base import BaseRepository


class OssLogRepository(BaseRepository[OssLog]):
    model = OssLog

    def create(
        self,
        *,
        log_name: str,
        agent_name: str | None,
        oss_path: str,
    ) -> OssLog:
        entity = OssLog(
            log_name=log_name,
            agent_name=agent_name,
            oss_path=oss_path,
            is_delete=0,
        )
        return self.add(entity)

    def list_by_agent_name(self, agent_name: str) -> list[OssLog]:
        """按 agent_name 查未删除记录，create_time 倒序。"""
        stmt = (
            select(OssLog)
            .where(
                OssLog.agent_name == agent_name,
                OssLog.is_delete == 0,
            )
            .order_by(OssLog.create_time.desc(), OssLog.id.desc())
        )
        return list(self.session.scalars(stmt).all())

    def list_all(self) -> list[OssLog]:
        """查询全部未删除记录，create_time 倒序。"""
        stmt = (
            select(OssLog)
            .where(OssLog.is_delete == 0)
            .order_by(OssLog.create_time.desc(), OssLog.id.desc())
        )
        return list(self.session.scalars(stmt).all())

    def list_agent_names(self) -> list[str]:
        """返回有日志记录的 Agent 名称。"""
        stmt = (
            select(distinct(OssLog.agent_name))
            .where(
                OssLog.is_delete == 0,
                OssLog.agent_name.is_not(None),
                OssLog.agent_name != "",
            )
            .order_by(OssLog.agent_name)
        )
        return [name for name in self.session.scalars(stmt).all() if name]