于小丘
b12aad4e73
- Add file deduplication mechanism based on PhysicalFile reference counting. - Implement chunked upload support for large files with resumable uploads. - Update sharing page to automatically render README and preview content. - Integrate Redis for caching and token storage (optional). - Refactor project structure to include new models for download tasks, nodes, and tasks. - Introduce user filtering parameters for admin user management. - Add CORS middleware for handling cross-origin requests. - Improve error messages for authentication failures. - Update user model to include two-factor authentication key management. - Enhance API documentation and response models for clarity. - Implement admin checks for user management and permissions.
97 lines
3.6 KiB
Python
97 lines
3.6 KiB
Python
"""
|
|
FastAPI 依赖注入
|
|
|
|
包含 HTTP 端点的通用依赖：
|
|
- SessionDep: 数据库会话依赖
|
|
- TimeFilterRequestDep: 时间筛选查询依赖（用于 count 等统计接口）
|
|
- TableViewRequestDep: 分页排序查询依赖（包含时间筛选 + 分页排序）
|
|
- UserFilterParamsDep: 用户筛选参数依赖（用于管理员用户列表）
|
|
"""
|
|
from datetime import datetime
|
|
from typing import Annotated, Literal, TypeAlias
|
|
from uuid import UUID
|
|
|
|
from fastapi import Depends, Query
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
from models.database import get_session
|
|
from models.mixin import TimeFilterRequest, TableViewRequest
|
|
from models.user import UserFilterParams, UserStatus
|
|
|
|
|
|
# --- 数据库会话依赖 ---
|
|
|
|
SessionDep: TypeAlias = Annotated[AsyncSession, Depends(get_session)]
|
|
"""数据库会话依赖，用于路由函数中获取数据库会话"""
|
|
|
|
|
|
# --- 时间筛选依赖 ---
|
|
|
|
async def _get_time_filter_queries(
|
|
created_after_datetime: Annotated[datetime | None, Query()] = None,
|
|
created_before_datetime: Annotated[datetime | None, Query()] = None,
|
|
updated_after_datetime: Annotated[datetime | None, Query()] = None,
|
|
updated_before_datetime: Annotated[datetime | None, Query()] = None,
|
|
) -> TimeFilterRequest:
|
|
"""解析时间筛选查询参数"""
|
|
return TimeFilterRequest(
|
|
created_after_datetime=created_after_datetime,
|
|
created_before_datetime=created_before_datetime,
|
|
updated_after_datetime=updated_after_datetime,
|
|
updated_before_datetime=updated_before_datetime,
|
|
)
|
|
|
|
|
|
TimeFilterRequestDep: TypeAlias = Annotated[TimeFilterRequest, Depends(_get_time_filter_queries)]
|
|
"""获取时间筛选参数的依赖（用于 count 等统计接口）"""
|
|
|
|
|
|
# --- 分页排序依赖 ---
|
|
|
|
async def _get_table_view_queries(
|
|
offset: Annotated[int | None, Query(ge=0)] = 0,
|
|
limit: Annotated[int | None, Query(ge=1, le=100)] = 20,
|
|
desc: bool | None = True,
|
|
order: Literal["created_at", "updated_at"] | None = "created_at",
|
|
created_after_datetime: Annotated[datetime | None, Query()] = None,
|
|
created_before_datetime: Annotated[datetime | None, Query()] = None,
|
|
updated_after_datetime: Annotated[datetime | None, Query()] = None,
|
|
updated_before_datetime: Annotated[datetime | None, Query()] = None,
|
|
) -> TableViewRequest:
|
|
"""解析分页排序和时间筛选查询参数"""
|
|
return TableViewRequest(
|
|
offset=offset,
|
|
limit=limit,
|
|
desc=desc,
|
|
order=order,
|
|
created_after_datetime=created_after_datetime,
|
|
created_before_datetime=created_before_datetime,
|
|
updated_after_datetime=updated_after_datetime,
|
|
updated_before_datetime=updated_before_datetime,
|
|
)
|
|
|
|
|
|
TableViewRequestDep: TypeAlias = Annotated[TableViewRequest, Depends(_get_table_view_queries)]
|
|
"""获取分页排序和时间筛选参数的依赖"""
|
|
|
|
|
|
# --- 用户筛选依赖 ---
|
|
|
|
async def _get_user_filter_params(
|
|
group_id: Annotated[UUID | None, Query(description="按用户组UUID筛选")] = None,
|
|
username: Annotated[str | None, Query(max_length=50, description="按用户名模糊搜索")] = None,
|
|
nickname: Annotated[str | None, Query(max_length=50, description="按昵称模糊搜索")] = None,
|
|
status: Annotated[UserStatus | None, Query(description="按用户状态筛选")] = None,
|
|
) -> UserFilterParams:
|
|
"""解析用户过滤查询参数"""
|
|
return UserFilterParams(
|
|
group_id=group_id,
|
|
username_contains=username,
|
|
nickname_contains=nickname,
|
|
status=status,
|
|
)
|
|
|
|
|
|
UserFilterParamsDep: TypeAlias = Annotated[UserFilterParams, Depends(_get_user_filter_params)]
|
|
"""获取用户筛选参数的依赖（用于管理员用户列表）"""
|