144 lines
4.5 KiB
Python
144 lines
4.5 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .agent_message_service import (
|
|
build_agent_message_docs,
|
|
can_run_console_test,
|
|
get_template_by_slug,
|
|
run_console_template_test,
|
|
serialize_agent_template,
|
|
)
|
|
from .auth import current_user
|
|
from .database import get_db
|
|
from .models import AgentMessageTemplate, AgentTemplateSubscription, UserAccount
|
|
|
|
router = APIRouter(prefix="/api/v1/agent-templates", tags=["agent-templates"])
|
|
|
|
|
|
class AgentTemplateTestRequest(BaseModel):
|
|
to: str = Field(min_length=10, max_length=20)
|
|
customer_name: str = Field(min_length=1, max_length=120)
|
|
report_id: str | None = Field(default=None, max_length=120)
|
|
|
|
|
|
def _subscription_map(db: Session, user: UserAccount) -> set[str]:
|
|
rows = db.scalars(
|
|
select(AgentTemplateSubscription.template_id).where(AgentTemplateSubscription.user_id == user.id)
|
|
).all()
|
|
return set(rows)
|
|
|
|
|
|
@router.get("")
|
|
def list_agent_templates(
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> list[dict]:
|
|
subscribed_ids = _subscription_map(db, user)
|
|
templates = db.scalars(
|
|
select(AgentMessageTemplate)
|
|
.where(AgentMessageTemplate.is_active.is_(True))
|
|
.order_by(AgentMessageTemplate.created_at.asc())
|
|
).all()
|
|
return [
|
|
serialize_agent_template(template, subscribed=template.id in subscribed_ids)
|
|
for template in templates
|
|
]
|
|
|
|
|
|
@router.post("/{slug}/subscribe")
|
|
def subscribe_agent_template(
|
|
slug: str,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
template = get_template_by_slug(db, slug)
|
|
existing = db.scalar(
|
|
select(AgentTemplateSubscription).where(
|
|
AgentTemplateSubscription.user_id == user.id,
|
|
AgentTemplateSubscription.template_id == template.id,
|
|
)
|
|
)
|
|
if existing:
|
|
return serialize_agent_template(template, subscribed=True)
|
|
db.add(AgentTemplateSubscription(user_id=user.id, template_id=template.id))
|
|
db.commit()
|
|
return serialize_agent_template(template, subscribed=True)
|
|
|
|
|
|
@router.delete("/{slug}/subscribe")
|
|
def unsubscribe_agent_template(
|
|
slug: str,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
template = get_template_by_slug(db, slug)
|
|
subscription = db.scalar(
|
|
select(AgentTemplateSubscription).where(
|
|
AgentTemplateSubscription.user_id == user.id,
|
|
AgentTemplateSubscription.template_id == template.id,
|
|
)
|
|
)
|
|
if subscription:
|
|
db.delete(subscription)
|
|
db.commit()
|
|
return serialize_agent_template(template, subscribed=False)
|
|
|
|
|
|
@router.get("/{slug}/docs")
|
|
def agent_template_docs(
|
|
slug: str,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
template = get_template_by_slug(db, slug)
|
|
subscription = db.scalar(
|
|
select(AgentTemplateSubscription).where(
|
|
AgentTemplateSubscription.user_id == user.id,
|
|
AgentTemplateSubscription.template_id == template.id,
|
|
)
|
|
)
|
|
if not subscription:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Subscribe to this template to view API documentation",
|
|
)
|
|
return build_agent_message_docs(template)
|
|
|
|
|
|
@router.get("/{slug}/test-status")
|
|
def agent_template_test_status(
|
|
slug: str,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
template = get_template_by_slug(db, slug)
|
|
require = db.scalar(
|
|
select(AgentTemplateSubscription).where(
|
|
AgentTemplateSubscription.user_id == user.id,
|
|
AgentTemplateSubscription.template_id == template.id,
|
|
)
|
|
)
|
|
if not require:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Subscribe first")
|
|
status_payload = can_run_console_test(db, user, template)
|
|
return {"template": template.slug, **status_payload}
|
|
|
|
|
|
@router.post("/{slug}/test")
|
|
async def test_agent_template(
|
|
slug: str,
|
|
payload: AgentTemplateTestRequest,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
template = get_template_by_slug(db, slug)
|
|
return await run_console_template_test(
|
|
db,
|
|
user,
|
|
template,
|
|
to=payload.to,
|
|
customer_name=payload.customer_name,
|
|
report_id=payload.report_id,
|
|
)
|