JiuContinent
d7db84571d
feat: 添加飞书用户模块和订阅功能支持 - 新增feishu_users模块用于处理飞书用户身份验证和权限管理 - 新增subscriptions模块用于处理订阅相关功能 - 新增personalization模块用于个性化服务 - 在alembic迁移配置中注册新的模型模块 - 在API路由器中添加feishu_users和subscriptions路由 - 实现事件调度服务的改进，包括错误处理和状态更新优化 - 添加飞书命令处理的权限检查机制 - 实现飞书应用票据事件处理 - 改进审计日志记录功能 ```
114 lines
3.4 KiB
Python
114 lines
3.4 KiB
Python
from typing import Any
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import select
|
|
|
|
from app.core.http.pagination import bounded_limit
|
|
from app.core.utils.time import utc_now
|
|
from app.modules.business.models import (
|
|
ReportPushRun,
|
|
)
|
|
from app.modules.business.service import serialize_model
|
|
from app.modules.reports.constants import (
|
|
ReportErrorDetail,
|
|
ReportPushStatus,
|
|
)
|
|
|
|
from app.modules.reports.services.common import _json_safe, _next_code
|
|
|
|
|
|
class ReportPushRunMixin:
|
|
def create_push_run(
|
|
self,
|
|
report_type: str,
|
|
title: str | None,
|
|
receive_id: str | None,
|
|
receive_id_type: str,
|
|
actor: str,
|
|
status: str = ReportPushStatus.PENDING,
|
|
task_id: str | None = None,
|
|
idempotency_key: str | None = None,
|
|
) -> ReportPushRun:
|
|
if idempotency_key:
|
|
existing = self.db.execute(
|
|
select(ReportPushRun).where(
|
|
ReportPushRun.idempotency_key == idempotency_key
|
|
)
|
|
).scalar_one_or_none()
|
|
if existing is not None:
|
|
return existing
|
|
record = ReportPushRun(
|
|
code=_next_code("PUSH"),
|
|
report_type=report_type,
|
|
title=title,
|
|
receive_id=receive_id,
|
|
receive_id_type=receive_id_type,
|
|
status=status,
|
|
task_id=task_id,
|
|
actor=actor,
|
|
queued_at=utc_now(),
|
|
idempotency_key=idempotency_key,
|
|
)
|
|
self.db.add(record)
|
|
self.db.commit()
|
|
self.db.refresh(record)
|
|
return record
|
|
|
|
def update_push_run(
|
|
self,
|
|
code: str,
|
|
status: str,
|
|
task_id: str | None = None,
|
|
provider_response: dict[str, Any] | None = None,
|
|
error_message: str | None = None,
|
|
sent: bool = False,
|
|
commit: bool = True,
|
|
) -> ReportPushRun:
|
|
record = self._get_push_run(code)
|
|
record.status = status
|
|
if task_id is not None:
|
|
record.task_id = task_id
|
|
if provider_response is not None:
|
|
record.provider_response = _json_safe(provider_response)
|
|
record.error_message = error_message
|
|
if sent:
|
|
record.sent_at = utc_now()
|
|
if commit:
|
|
self.db.commit()
|
|
self.db.refresh(record)
|
|
else:
|
|
self.db.flush()
|
|
return record
|
|
|
|
def list_push_runs(
|
|
self,
|
|
status_filter: str | None = None,
|
|
limit: int = 100,
|
|
) -> list[dict[str, Any]]:
|
|
stmt = select(ReportPushRun).order_by(ReportPushRun.id.desc()).limit(
|
|
bounded_limit(limit)
|
|
)
|
|
if status_filter:
|
|
stmt = (
|
|
select(ReportPushRun)
|
|
.where(ReportPushRun.status == status_filter)
|
|
.order_by(ReportPushRun.id.desc())
|
|
.limit(bounded_limit(limit))
|
|
)
|
|
return [serialize_model(item) for item in self.db.execute(stmt).scalars()]
|
|
|
|
def get_push_run(self, code: str) -> dict[str, Any]:
|
|
return serialize_model(self._get_push_run(code))
|
|
|
|
def _get_push_run(self, code: str) -> ReportPushRun:
|
|
|
|
record = self.db.execute(
|
|
select(ReportPushRun).where(ReportPushRun.code == code)
|
|
).scalar_one_or_none()
|
|
if record is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=ReportErrorDetail.PUSH_RUN_NOT_FOUND,
|
|
)
|
|
return record
|