mirror of
https://github.com/csunny/DB-GPT.git
synced 2025-08-04 01:50:08 +00:00
b124ecc10b
Co-authored-by: 夏姜 <wenfengjiang.jwf@digital-engine.com> Co-authored-by: aries_ckt <916701291@qq.com> Co-authored-by: wb-lh513319 <wb-lh513319@alibaba-inc.com> Co-authored-by: csunny <cfqsunny@163.com>
94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
from datetime import datetime
|
|
from typing import Any, Dict, Optional, Union
|
|
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import BigInteger, Column, DateTime, String
|
|
|
|
from dbgpt.storage.metadata import BaseDao, Model
|
|
|
|
|
|
class SettingsEntity(Model):
|
|
"""Settings entity"""
|
|
|
|
__tablename__ = "settings"
|
|
id = Column(BigInteger, primary_key=True, autoincrement=True, comment="Primary Key")
|
|
gmt_create = Column(DateTime, default=datetime.now, comment="Creation time")
|
|
gmt_modify = Column(
|
|
DateTime,
|
|
default=datetime.now,
|
|
comment="Modification time",
|
|
)
|
|
setting_key = Column(String(32), nullable=False, comment="Configuration key")
|
|
setting_value = Column(String(255), nullable=True, comment="Configuration value")
|
|
description = Column(
|
|
String(255), nullable=True, comment="Configuration description"
|
|
)
|
|
|
|
def __repr__(self):
|
|
return (
|
|
f"SettingsEntity(id={self.id}, setting_key='{self.setting_key}', setting_value='{self.setting_value}', "
|
|
f"description='{self.description}', gmt_create='{self.gmt_create}', gmt_modify='{self.gmt_modify}')"
|
|
)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.id,
|
|
"setting_key": self.setting_key,
|
|
"setting_value": self.setting_value,
|
|
"description": self.description,
|
|
"gmt_create": self.gmt_create,
|
|
"gmt_modify": self.gmt_modify,
|
|
}
|
|
|
|
|
|
class SettingsRequest(BaseModel):
|
|
"""Settings request."""
|
|
|
|
setting_key: str = Field(..., description="The configuration key")
|
|
setting_value: Optional[str] = Field(None, description="The configuration value")
|
|
description: Optional[str] = Field(
|
|
None, description="The configuration description"
|
|
)
|
|
|
|
|
|
class SettingsResponse(BaseModel):
|
|
id: Optional[int] = Field(None, description="The primary id")
|
|
setting_key: str = Field(..., description="The configuration key")
|
|
setting_value: Optional[str] = Field(None, description="The configuration value")
|
|
description: Optional[str] = Field(
|
|
None, description="The configuration description"
|
|
)
|
|
|
|
|
|
class SettingsDao(BaseDao):
|
|
"""Settings dao."""
|
|
|
|
def from_request(self, request: SettingsRequest) -> Dict[str, Any]:
|
|
request_dict = (
|
|
request.dict() if isinstance(request, SettingsRequest) else request
|
|
)
|
|
entity = SettingsEntity(**request_dict)
|
|
return entity
|
|
|
|
def to_response(self, entity: SettingsEntity) -> SettingsResponse:
|
|
return SettingsResponse(
|
|
id=entity.id,
|
|
setting_key=entity.setting_key,
|
|
setting_value=entity.setting_value,
|
|
description=entity.description,
|
|
)
|
|
|
|
def to_request(self, entity: SettingsEntity) -> SettingsRequest:
|
|
return SettingsRequest(
|
|
setting_key=entity.setting_key,
|
|
setting_value=entity.setting_value,
|
|
description=entity.description,
|
|
)
|
|
|
|
def from_response(self, response: SettingsResponse) -> Dict[str, Any]:
|
|
response_dict = (
|
|
response.dict() if isinstance(response, SettingsResponse) else response
|
|
)
|
|
entity = SettingsEntity(**response_dict)
|
|
return entity
|