⬆️ upgrade Pydantic to V2

This commit is contained in:
xtaodada 2024-11-30 20:51:20 +08:00
parent 112b2e92d8
commit bf5b153001
Signed by: xtaodada
GPG Key ID: 4CBB3F4FA8C85659
4 changed files with 16 additions and 34 deletions

View File

@ -1,13 +1,12 @@
import enum
from pydantic_settings import BaseSettings, SettingsConfigDict
try:
import ujson as jsonlib
except ImportError:
import json as jsonlib
from pydantic import BaseSettings
__all__ = ("RegionEnum", "Settings")
__all__ = ("RegionEnum", "Settings", "SettingsConfigDict")
class RegionEnum(int, enum.Enum):
@ -20,10 +19,5 @@ class RegionEnum(int, enum.Enum):
class Settings(BaseSettings):
def __new__(cls, *args, **kwargs):
cls.update_forward_refs()
cls.model_rebuild()
return super(Settings, cls).__new__(cls) # pylint: disable=E1120
class Config(BaseSettings.Config):
case_sensitive = False
json_loads = jsonlib.loads
json_dumps = jsonlib.dumps

View File

@ -4,6 +4,7 @@ from typing import List, Optional, Set
import dotenv
from pydantic import AnyUrl, Field
from pydantic_settings import SettingsConfigDict
from gram_core.basemodel import Settings
from utils.const import PROJECT_ROOT
@ -29,8 +30,7 @@ class DatabaseConfig(Settings):
password: Optional[str] = None
database: Optional[str] = None
class Config(Settings.Config):
env_prefix = "db_"
model_config = SettingsConfigDict(env_prefix="db_")
class InfluxDBConfig(Settings):
@ -39,18 +39,16 @@ class InfluxDBConfig(Settings):
token: Optional[str] = None
org: Optional[str] = None
class Config(Settings.Config):
env_prefix = "influxdb_"
model_config = SettingsConfigDict(env_prefix="influxdb_")
class RedisConfig(Settings):
host: str = "127.0.0.1"
port: int = 6379
database: int = Field(default=0, env="redis_db")
database: int = Field(default=0, validation_alias="redis_db")
password: Optional[str] = None
class Config(Settings.Config):
env_prefix = "redis_"
model_config = SettingsConfigDict(env_prefix="redis_")
class LoggerConfig(Settings):
@ -65,8 +63,7 @@ class LoggerConfig(Settings):
locals_max_depth: Optional[NaturalNumber] = None
filtered_names: List[str] = ["uvicorn"]
class Config(Settings.Config):
env_prefix = "logger_"
model_config = SettingsConfigDict(env_prefix="logger_")
class MTProtoConfig(Settings):
@ -82,8 +79,7 @@ class WebServerConfig(Settings):
host: str = "localhost"
port: int = 8080
class Config(Settings.Config):
env_prefix = "web_"
model_config = SettingsConfigDict(env_prefix="web_")
class ErrorConfig(Settings):
@ -93,8 +89,7 @@ class ErrorConfig(Settings):
sentry_dsn: str = ""
notification_chat_id: Optional[str] = None
class Config(Settings.Config):
env_prefix = "error_"
model_config = SettingsConfigDict(env_prefix="error_")
class ReloadConfig(Settings):
@ -103,8 +98,7 @@ class ReloadConfig(Settings):
include: List[str] = []
exclude: List[str] = []
class Config(Settings.Config):
env_prefix = "reload_"
model_config = SettingsConfigDict(env_prefix="reload_")
class NoticeConfig(Settings):
@ -117,8 +111,7 @@ class NoticeConfig(Settings):
"""拒绝加入群聊"""
quit_status: str = f"{bot_name}不想进去!不是旅行者的邀请!"
class Config(Settings.Config):
env_prefix = "notice_"
model_config = SettingsConfigDict(env_prefix="notice_")
class BotConfig(Settings):
@ -137,8 +130,7 @@ class BotConfig(Settings):
webhook_url: str = "http://127.0.0.1:8080/telegram"
"""webhook url"""
class Config(Settings.Config):
env_prefix = "bot_"
model_config = SettingsConfigDict(env_prefix="bot_")
class ApplicationConfig(Settings):

View File

@ -1,7 +1,7 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, BaseSettings
from pydantic import BaseModel
from sqlalchemy import TypeDecorator
from sqlmodel import Boolean, Column, Enum, Field, SQLModel, Integer, Index, BigInteger, VARCHAR, func, DateTime
@ -33,10 +33,6 @@ class PlayersDataBase(Player, table=True):
class ExtraPlayerInfo(BaseModel):
class Config(BaseSettings.Config):
frozen = False
json_loads = jsonlib.loads
json_dumps = jsonlib.dumps
waifu_id: Optional[int] = None
level: Optional[int] = None

View File

@ -163,7 +163,7 @@ class TemplatePreviewer(BaseService, load=application_config.webserver.enable an
async def get_preview_url(self, template: str, data: dict):
"""获取预览 URL"""
components = urlsplit(application_config.webserver.url)
components = urlsplit(str(application_config.webserver.url))
path = urljoin("/preview/", template)
query = {}