mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-22 13:05:18 +00:00
zhangtao
b0ae6af3b1
- Added *.pyc and *.pyo to .gitignore to prevent compiled Python files from being tracked. - Improved docstrings in various modules, providing clearer descriptions of functions, parameters, and return values to enhance code readability and maintainability.
204 lines
6.5 KiB
Python
204 lines
6.5 KiB
Python
from datetime import datetime
|
|
from urllib.parse import urlparse
|
|
|
|
from fastapi import Query
|
|
from pydantic import (
|
|
BaseModel,
|
|
ConfigDict,
|
|
Field,
|
|
field_validator,
|
|
model_validator,
|
|
)
|
|
|
|
from app.common.enums import QueueEnum
|
|
|
|
|
|
class ResourceItemSchema(BaseModel):
|
|
"""资源项目模型"""
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
name: str = Field(..., description="文件名")
|
|
file_url: str = Field(..., description="文件URL路径")
|
|
relative_path: str = Field(..., description="相对路径")
|
|
is_file: bool = Field(..., description="是否为文件")
|
|
is_dir: bool = Field(..., description="是否为目录")
|
|
size: int | None = Field(None, description="文件大小(字节)")
|
|
created_time: datetime | None = Field(None, description="创建时间")
|
|
modified_time: datetime | None = Field(None, description="修改时间")
|
|
is_hidden: bool = Field(False, description="是否为隐藏文件")
|
|
|
|
@field_validator("file_url")
|
|
@classmethod
|
|
def _validate_file_url(cls, v: str) -> str:
|
|
v = v.strip()
|
|
parsed = urlparse(v)
|
|
# 允许相对路径（以 / 开头）和完整的 http/https URL
|
|
if parsed.scheme and parsed.scheme not in ("http", "https"):
|
|
raise ValueError("文件URL必须为 http/https 或相对路径")
|
|
return v
|
|
|
|
@field_validator("relative_path")
|
|
@classmethod
|
|
def _validate_relative_path(cls, v: str) -> str:
|
|
v = v.strip()
|
|
if ".." in v or v.startswith("\\"):
|
|
raise ValueError("相对路径包含不安全字符")
|
|
return v
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_flags(self):
|
|
if self.is_file and self.is_dir:
|
|
raise ValueError("不能同时为文件和目录")
|
|
if not self.is_file and not self.is_dir:
|
|
raise ValueError("必须是文件或目录之一")
|
|
# 根据名称自动修正隐藏标记
|
|
self.is_hidden = self.name.startswith(".")
|
|
return self
|
|
|
|
|
|
class ResourceDirectorySchema(BaseModel):
|
|
"""资源目录模型"""
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
path: str = Field(..., description="目录路径")
|
|
name: str = Field(..., description="目录名称")
|
|
items: list[ResourceItemSchema] = Field(default_factory=list, description="目录项")
|
|
total_files: int = Field(0, description="文件总数")
|
|
total_dirs: int = Field(0, description="目录总数")
|
|
total_size: int = Field(0, description="总大小")
|
|
|
|
|
|
class ResourceUploadSchema(BaseModel):
|
|
"""资源上传响应模型"""
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
filename: str = Field(..., description="文件名")
|
|
file_url: str = Field(..., description="访问URL")
|
|
file_size: int = Field(..., description="文件大小")
|
|
upload_time: datetime = Field(..., description="上传时间")
|
|
|
|
|
|
class ResourceMoveSchema(BaseModel):
|
|
"""资源移动模型"""
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
source_path: str = Field(..., description="源路径")
|
|
target_path: str = Field(..., description="目标路径")
|
|
overwrite: bool = Field(False, description="是否覆盖")
|
|
|
|
@field_validator("source_path", "target_path")
|
|
@classmethod
|
|
def validate_paths(cls, value: str):
|
|
"""
|
|
校验移动/复制涉及的源路径与目标路径非空并去首尾空格。
|
|
|
|
参数:
|
|
- value (str): 路径字段当前值。
|
|
|
|
返回:
|
|
- str: 去空格后的路径。
|
|
|
|
异常:
|
|
- ValueError: 路径为空时抛出。
|
|
"""
|
|
if not value or len(value.strip()) == 0:
|
|
raise ValueError("路径不能为空")
|
|
return value.strip()
|
|
|
|
|
|
class ResourceCopySchema(ResourceMoveSchema):
|
|
"""资源复制模型"""
|
|
|
|
|
|
class ResourceRenameSchema(BaseModel):
|
|
"""资源重命名模型"""
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
old_path: str = Field(..., description="原路径")
|
|
new_name: str = Field(..., description="新名称")
|
|
|
|
@field_validator("old_path", "new_name")
|
|
@classmethod
|
|
def validate_inputs(cls, value: str):
|
|
"""
|
|
校验重命名所需的原路径与新名称非空并去首尾空格。
|
|
|
|
参数:
|
|
- value (str): 字段当前值。
|
|
|
|
返回:
|
|
- str: 去空格后的值。
|
|
|
|
异常:
|
|
- ValueError: 值为空时抛出。
|
|
"""
|
|
if not value or len(value.strip()) == 0:
|
|
raise ValueError("参数不能为空")
|
|
return value.strip()
|
|
|
|
@field_validator("new_name")
|
|
@classmethod
|
|
def _validate_new_name(cls, v: str) -> str:
|
|
v = v.strip()
|
|
if ".." in v or "/" in v or "\\" in v:
|
|
raise ValueError("新名称包含不安全字符")
|
|
return v
|
|
|
|
|
|
class ResourceCreateDirSchema(BaseModel):
|
|
"""创建目录模型"""
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
parent_path: str = Field(..., description="父目录路径")
|
|
dir_name: str = Field(..., description="目录名称", max_length=255)
|
|
|
|
@field_validator("parent_path", "dir_name")
|
|
@classmethod
|
|
def validate_inputs(cls, value: str, info):
|
|
"""
|
|
校验创建目录的父路径与目录名，防止路径遍历等不安全输入。
|
|
|
|
参数:
|
|
- value (str): 当前字段值。
|
|
- info: Pydantic 校验上下文（含 `field_name`）。
|
|
|
|
返回:
|
|
- str: 规范化后的字段值。
|
|
|
|
异常:
|
|
- ValueError: 含不安全字符或目录名为空时抛出。
|
|
"""
|
|
# 对于parent_path允许为空字符串（表示根目录）或 '/'，其他情况必须非空
|
|
if info.field_name == "parent_path":
|
|
# 对于parent_path仍然严格检查路径遍历
|
|
if ".." in value or value.startswith("\\"):
|
|
raise ValueError("参数包含不安全字符")
|
|
else: # 对于dir_name仍然严格检查
|
|
if not value or len(value.strip()) == 0:
|
|
raise ValueError("参数不能为空")
|
|
if ".." in value or value.startswith(("/", "\\")):
|
|
raise ValueError("参数包含不安全字符")
|
|
return value.strip()
|
|
|
|
|
|
class ResourceSearchQueryParam:
|
|
"""资源搜索查询参数"""
|
|
|
|
def __init__(
|
|
self,
|
|
name: str | None = Query(None, description="搜索关键词"),
|
|
path: str | None = Query(None, description="目录路径"),
|
|
) -> None:
|
|
|
|
# 模糊查询字段
|
|
self.name = (QueueEnum.like.value, name) if name else None
|
|
|
|
# 精确查询字段
|
|
self.path = path
|