from typing import Literal, Optional from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import joinedload from ..config import get_settings from ..database import get_db from ..models import App, AppCategoryMap, AppVersion, Category from ..schemas import ( AppDetail, AppListResponse, AppSummary, DownloadInfo, ErrorResponse, ManifestOut, VersionInfo, ) router = APIRouter(prefix="/api/v1", tags=["Apps"]) AppType = Literal["system", "community", "external"] SortType = Literal["newest", "popular"] def _build_base_query(): return ( select(App) .options( joinedload(App.developer), joinedload(App.categories), joinedload(App.current_version), ) .where(App.status == "published") ) def _apply_filters( stmt, category: Optional[str], app_type: Optional[str], search: Optional[str], featured: Optional[bool], ): if category: stmt = ( stmt.join(AppCategoryMap, App.id == AppCategoryMap.app_id) .join(Category, Category.id == AppCategoryMap.category_id) .where(Category.slug == category) ) if app_type: stmt = stmt.where(App.type == app_type) if search: pattern = f"%{search}%" stmt = stmt.where( App.name.ilike(pattern) | App.short_description.ilike(pattern) ) if featured is not None: stmt = stmt.where(App.is_featured == featured) return stmt def _apply_sort(stmt, sort: str): if sort == "popular": return stmt.order_by(App.download_count.desc(), App.publish_time.desc()) return stmt.order_by(App.publish_time.desc(), App.id.desc()) def _get_published_version(app: App) -> Optional[AppVersion]: """Get published version, falling back to latest published if current is not published.""" if app.current_version and app.current_version.status == "published": return app.current_version return None def _to_version_info(version: Optional[AppVersion]) -> Optional[VersionInfo]: if not version or version.status != "published": return None return VersionInfo( version=version.version, release_notes=version.release_notes, min_client_version=version.min_client_version, published_at=version.published_at, size_bytes=version.size_bytes, ) def _to_app_summary(app: App) -> AppSummary: return AppSummary( app_uid=app.app_uid, name=app.name, type=app.type, icon_url=app.icon_url, short_description=app.short_description, categories=app.categories, developer=app.developer, current_version=_to_version_info(app.current_version), download_count=app.download_count, is_featured=app.is_featured, publish_time=app.publish_time, ) @router.get("/apps", response_model=AppListResponse) async def list_apps( category: Optional[str] = Query(None, description="Filter by category slug"), type: Optional[AppType] = Query(None, description="Filter by app type"), search: Optional[str] = Query(None, min_length=1, description="Search keyword"), featured: Optional[bool] = Query(None, description="Filter featured apps only"), sort: SortType = Query("newest", description="Sort order"), page: int = Query(1, ge=1, description="Page number"), page_size: Optional[int] = Query(None, ge=1, le=100, description="Items per page"), db: AsyncSession = Depends(get_db), ) -> AppListResponse: settings = get_settings() limit = min(page_size or settings.default_page_size, settings.max_page_size) offset = (page - 1) * limit base_stmt = _build_base_query().distinct() filtered_stmt = _apply_filters(base_stmt, category, type, search, featured) count_base = select(func.count(func.distinct(App.id))).where(App.status == "published") count_stmt = _apply_filters(count_base, category, type, search, featured) total_result = await db.execute(count_stmt) total = total_result.scalar_one() or 0 query_stmt = _apply_sort(filtered_stmt, sort).offset(offset).limit(limit) result = await db.execute(query_stmt) apps = result.scalars().unique().all() return AppListResponse( items=[_to_app_summary(app) for app in apps], total=total, page=page, page_size=limit, ) @router.get( "/apps/{app_uid}", response_model=AppDetail, responses={404: {"model": ErrorResponse}}, ) async def get_app_detail(app_uid: str, db: AsyncSession = Depends(get_db)) -> AppDetail: stmt = _build_base_query().where(App.app_uid == app_uid) result = await db.execute(stmt) app = result.scalars().unique().one_or_none() if not app: raise HTTPException(status_code=404, detail="App not found") manifest = None if app.current_version and app.current_version.manifest: m = app.current_version.manifest manifest = ManifestOut( pages=m.get("pages"), commands=m.get("commands"), permissions=m.get("permissions"), ) return AppDetail( app_uid=app.app_uid, name=app.name, type=app.type, icon_url=app.icon_url, short_description=app.short_description, description=app.description, homepage_url=app.homepage_url, categories=app.categories, developer=app.developer, current_version=_to_version_info(app.current_version), download_count=app.download_count, is_featured=app.is_featured, publish_time=app.publish_time, manifest=manifest, ) @router.get( "/apps/{app_uid}/download", response_model=DownloadInfo, responses={404: {"model": ErrorResponse}}, ) async def get_download_info(app_uid: str, db: AsyncSession = Depends(get_db)) -> DownloadInfo: stmt = _build_base_query().where(App.app_uid == app_uid) result = await db.execute(stmt) app = result.scalars().unique().one_or_none() if not app: raise HTTPException(status_code=404, detail="App not found") version = app.current_version if not version or version.status != "published": version_stmt = ( select(AppVersion) .where(AppVersion.app_id == app.id, AppVersion.status == "published") .order_by(AppVersion.published_at.desc()) .limit(1) ) version_result = await db.execute(version_stmt) version = version_result.scalar_one_or_none() if not version: raise HTTPException(status_code=404, detail="No published version available") return DownloadInfo( app_uid=app.app_uid, version=version.version, url=version.storage_url, checksum_sha256=version.checksum_sha256, size_bytes=version.size_bytes, )