zc
5 months ago
parent
41f24b5bf6
commit
810e28ecba
@ -0,0 +1,141 @@ |
||||
from datetime import datetime |
||||
from typing import Optional, List, Dict, Any, Union |
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator |
||||
from sqlalchemy import Column, Integer, String, DateTime, Boolean, text |
||||
from dbgpt._private.pydantic import BaseModel as DBGPTBaseModel |
||||
from dbgpt.storage.metadata import Model |
||||
|
||||
|
||||
|
||||
class TypeBase(DBGPTBaseModel): |
||||
"""分类基础模型""" |
||||
model_config = ConfigDict(arbitrary_types_allowed=True) |
||||
|
||||
id: Optional[int] = Field(None, description="主键ID") |
||||
type_label: Optional[str] = Field(None, description="分类标签", max_length=100) |
||||
type_value: Optional[str] = Field(None, description="分类键值", max_length=100) |
||||
remark: Optional[str] = Field(None, description="备注", max_length=500) |
||||
creator: Optional[str] = Field("", description="创建者", max_length=64) |
||||
create_time: Optional[datetime] = Field(None, description="创建时间") |
||||
updater: Optional[str] = Field("", description="更新者", max_length=64) |
||||
update_time: Optional[datetime] = Field(None, description="更新时间") |
||||
deleted: Optional[bool] = Field(False, description="是否删除") |
||||
|
||||
def to_dict(self) -> Dict[str, Any]: |
||||
"""转换为字典（自动过滤None值）""" |
||||
return {k: v for k, v in self.model_dump().items() if v is not None} |
||||
|
||||
@classmethod |
||||
def from_dict(cls, data: Dict[str, Any]) -> "TypeBase": |
||||
"""从字典创建对象""" |
||||
return cls(**data) |
||||
|
||||
|
||||
class TypeCreate(TypeBase): |
||||
|
||||
"""分类创建模型""" |
||||
|
||||
@field_validator('type_label') |
||||
def validate_type_label(cls, v): |
||||
if not v: |
||||
raise ValueError("分类标签不能为空") |
||||
if not v or len(v.strip()) < 2: |
||||
raise ValueError("分类标签至少需要2个字符") |
||||
return v |
||||
|
||||
@field_validator('type_value') |
||||
def validate_type_value(cls, v): |
||||
if not v: |
||||
raise ValueError("分类键值不能为空") |
||||
if not v or len(v.strip()) < 2: |
||||
raise ValueError("分类键值至少需要2个字符") |
||||
return v |
||||
|
||||
|
||||
class TypeUpdate(TypeBase): |
||||
"""角色更新模型""" |
||||
|
||||
type_label: Optional[str] = Field(None, description="分类标签", max_length=100) |
||||
type_value: Optional[str] = Field(None, description="分类键值", max_length=100) |
||||
|
||||
@field_validator('type_label', 'type_value', mode='before') |
||||
def empty_str_to_none(cls, v: Optional[str]) -> Optional[str]: |
||||
"""将空字符串转为None（便于部分更新）""" |
||||
return v if v and v.strip() else None |
||||
|
||||
|
||||
class TypeQuery(TypeBase): |
||||
"""角色查询模型""" |
||||
page_size: int = Field(100, ge=1, le=1000, description="每页数量") |
||||
page: int = Field(1, ge=1, description="页码") |
||||
type_label: Optional[str] = Field(None, description="分类标签模糊查询") |
||||
|
||||
|
||||
class TypeResponse(DBGPTBaseModel): |
||||
"""角色分页响应模型""" |
||||
total_count: int = Field(0, description="总记录数") |
||||
total_page: int = Field(0, description="总页数") |
||||
current_page: int = Field(1, description="当前页") |
||||
types: List[TypeBase] = Field(default_factory=list, description="分类列表") |
||||
|
||||
|
||||
class TypeEntity(Model): |
||||
"""角色数据库实体""" |
||||
__tablename__ = "sys_type" |
||||
__table_args__ = { |
||||
"comment": "分类管理表", |
||||
"mysql_charset": "utf8mb4", |
||||
"mysql_collate": "utf8mb4_unicode_ci" |
||||
} |
||||
|
||||
id = Column( |
||||
Integer, |
||||
primary_key=True, |
||||
autoincrement=True, |
||||
comment="主键ID" |
||||
) |
||||
name = Column( |
||||
String(100), |
||||
nullable=False, |
||||
comment="分类标签" |
||||
) |
||||
code = Column( |
||||
String(100), |
||||
nullable=False, |
||||
comment="分类键值" |
||||
) |
||||
remark = Column( |
||||
String(500), |
||||
comment="备注" |
||||
) |
||||
creator = Column( |
||||
String(64), |
||||
server_default=text("''"), |
||||
comment="创建者" |
||||
) |
||||
create_time = Column( |
||||
DateTime, |
||||
server_default=text("CURRENT_TIMESTAMP"), |
||||
comment="创建时间" |
||||
) |
||||
updater = Column( |
||||
String(64), |
||||
server_default=text("''"), |
||||
comment="更新者" |
||||
) |
||||
update_time = Column( |
||||
DateTime, |
||||
server_default=text("CURRENT_TIMESTAMP"), |
||||
comment="更新时间" |
||||
) |
||||
deleted = Column( |
||||
Integer, |
||||
server_default=text("0"), |
||||
nullable=False, |
||||
comment="是否删除" |
||||
) |
||||
|
||||
|
||||
# ==================== 类型提示 ==================== |
||||
TypeList = List[TypeBase] |
||||
TypeDict = Dict[str, Any] |
||||
@ -0,0 +1,172 @@ |
||||
from datetime import datetime, timedelta |
||||
from dbgpt.storage.metadata import BaseDao |
||||
from typing import Optional |
||||
from .type import TypeBase, TypeCreate, TypeUpdate, TypeQuery, TypeResponse,TypeEntity |
||||
from typing import Optional, List, Dict, Any |
||||
|
||||
typeEntity = TypeEntity |
||||
|
||||
|
||||
class TypeDao(BaseDao): |
||||
|
||||
|
||||
def select_type_details(self, type_id: int) -> Optional[TypeBase]: |
||||
""" |
||||
根据条件查询单个角色 |
||||
""" |
||||
session = self.get_raw_session() |
||||
try: |
||||
type_obj = session.query(TypeEntity).filter( |
||||
TypeEntity.deleted == 0, |
||||
TypeEntity.id == type_id |
||||
).first() |
||||
return TypeBase.from_dict(type_obj.__dict__) if type_obj else None |
||||
finally: |
||||
session.close() |
||||
|
||||
def select_type_page(self, type_query: TypeQuery): |
||||
"""分页查询角色列表""" |
||||
session = self.get_raw_session() |
||||
try: |
||||
type_qry = session.query(TypeEntity).filter(TypeEntity.deleted == 0) |
||||
|
||||
# 动态过滤条件 |
||||
if type_query.type_label: |
||||
type_qry = type_qry.filter(TypeEntity.name.like(f"%{type_query.type_label}%")) |
||||
|
||||
# 分页处理 |
||||
total_count = type_qry.count() |
||||
type_qry = type_qry.order_by(TypeEntity.create_time.desc(), TypeEntity.id.desc()) |
||||
type_qry = type_qry.offset( |
||||
(type_query.page - 1) * type_query.page_size |
||||
).limit(type_query.page_size) |
||||
|
||||
results = type_qry.all() |
||||
finally: |
||||
session.close() |
||||
|
||||
# 构建响应 |
||||
type_resp = TypeResponse() |
||||
if results: |
||||
types = [] |
||||
for type_obj in results: |
||||
types.append( |
||||
TypeBase.from_dict({ |
||||
"id": type_obj.id, |
||||
"type_label": type_obj.type_label, |
||||
"type_value": type_obj.type_value, |
||||
"remark": type_obj.remark, |
||||
"creator": type_obj.creator, |
||||
"updater": type_obj.updater, |
||||
"create_time": type_obj.create_time, |
||||
"update_time": type_obj.update_time |
||||
}) |
||||
) |
||||
type_resp.total_count = total_count |
||||
type_resp.roles = types |
||||
type_resp.current_page = type_query.page |
||||
type_resp.total_page = (total_count + type_query.page_size - 1) // type_query.page_size |
||||
|
||||
return type_resp |
||||
|
||||
|
||||
def select_type_list(self) -> List[Dict[str, Any]]: |
||||
"""查询分类列表（仅返回id和type_label）""" |
||||
session = self.get_raw_session() |
||||
try: |
||||
# 固定查询条件：未删除的分类 |
||||
types = session.query( |
||||
TypeEntity.id, |
||||
TypeEntity.type_label |
||||
).filter( |
||||
TypeEntity.deleted == 0 |
||||
).all() |
||||
|
||||
# 直接构建字典列表返回 |
||||
return [{"id": r.id, "type_label": r.name} for r in types] |
||||
finally: |
||||
session.close() |
||||
|
||||
def create(self, type_obj: TypeCreate) -> bool: |
||||
"""创建分类""" |
||||
try: |
||||
with self.session() as session: |
||||
type_entity = TypeEntity( |
||||
type_label=type_obj.type_label, |
||||
type_value=type_obj.type_value, |
||||
remark=type_obj.remark or "", |
||||
creator=type_obj.creator or "1", |
||||
updater=type_obj.updater or "1", |
||||
deleted=0, |
||||
create_time=datetime.now(), |
||||
update_time=datetime.now() |
||||
) |
||||
session.add(type_entity) |
||||
session.commit() |
||||
return True |
||||
except Exception as e: |
||||
session.rollback() |
||||
raise RuntimeError(f"创建分类失败: {str(e)}") |
||||
|
||||
def update(self, type_obj: TypeUpdate) -> bool: |
||||
"""更新分类信息""" |
||||
with self.session() as session: |
||||
type_entity = session.query(TypeEntity).filter_by( |
||||
id=type_obj.id, |
||||
deleted=0 |
||||
).first() |
||||
|
||||
if not type_entity: |
||||
raise ValueError("分类不存在或已被删除") |
||||
|
||||
# 只更新非None字段 |
||||
if type_obj.type_label is not None: |
||||
type_entity.name = type_obj.type_label |
||||
if type_obj.type_value is not None: |
||||
type_entity.type_value = type_obj.type_value |
||||
if type_obj.remark is not None: |
||||
type_entity.remark = type_obj.remark |
||||
|
||||
type_entity.updater = type_obj.updater |
||||
type_entity.update_time = datetime.now() |
||||
|
||||
session.commit() |
||||
return True |
||||
|
||||
def delete_type(self, type_id: int, updater: int) -> bool: |
||||
"""软删除角色""" |
||||
session = self.get_raw_session() |
||||
try: |
||||
type_obj = session.query(TypeEntity).filter_by( |
||||
id=type_id, |
||||
deleted=0 |
||||
).first() |
||||
|
||||
if not type_obj: |
||||
raise ValueError("分类不存在或已被删除") |
||||
|
||||
type_obj.deleted = 1 |
||||
type_obj.updater = updater |
||||
type_obj.update_time = datetime.now() |
||||
|
||||
session.commit() |
||||
return True |
||||
except Exception as e: |
||||
session.rollback() |
||||
raise RuntimeError(f"删除分类失败: {str(e)}") |
||||
finally: |
||||
session.close() |
||||
|
||||
def check_type_name_exists(self, type_label: str, exclude_id: Optional[int] = None) -> bool: |
||||
""" 检查分类名称是否已存在 """ |
||||
session = self.get_raw_session() |
||||
try: |
||||
query = session.query(TypeEntity).filter( |
||||
TypeEntity.type_label == type_label, |
||||
TypeEntity.deleted == 0 |
||||
) |
||||
if exclude_id: |
||||
query = query.filter(TypeEntity.id != exclude_id) |
||||
return query.first() is not None |
||||
finally: |
||||
session.close() |
||||
Loading…
Reference in new issue