mdsc/common/config.py

87 lines
2.2 KiB
Python

import secrets
from typing import Optional
from typing import Dict
from typing import Any
from pydantic import BaseSettings
from pydantic import HttpUrl
from pydantic import validator
from pydantic import AnyUrl
class MysqlDsn(AnyUrl):
allowed_schemes = {'mysql'}
user_required = True
class RedisDsn(AnyUrl):
allowed_schemes = {'redis'}
@classmethod
def validate_parts(cls, parts: Dict[str, str]) -> Dict[str, str]:
defaults = {
'domain': 'localhost' if not (parts['ipv4'] or parts['ipv6']) else '',
'port': '6379'
}
for key, value in defaults.items():
if not parts[key]:
parts[key] = value
return super().validate_parts(parts)
class Settings(BaseSettings):
# JWT
SECRET_KEY: str = secrets.token_urlsafe(32)
ALGORITHM: str = 'HS256'
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 3
# Sentry
SENTRY_DSN: Optional[HttpUrl]
# 数据库
MYSQL_ADDRESS: str
MYSQL_USER: str
MYSQL_PASSWORD: str
MYSQL_DB: str
MYSQL_DATABASE_URL: Optional[MysqlDsn] = None
@validator("MYSQL_DATABASE_URL", pre=True)
def assemble_db_connection(cls, v: Optional[str], values: Dict[str, Any]) -> Any:
if isinstance(v, str):
return v
return MysqlDsn.build(
scheme="mysql",
user=values.get("MYSQL_USER"),
password=values.get("MYSQL_PASSWORD"),
host=values.get("MYSQL_ADDRESS"),
path=f"/{values.get('MYSQL_DB') or ''}",
)
# redis, ip + 端口
REDIS_ADDRESS: str
REDIS_URL: Optional[RedisDsn] = None
@validator("REDIS_URL", pre=True)
def assemble_redis_connection(cls, v: Optional[str], values: Dict[str, Any]) -> Any:
if isinstance(v, str):
return v
return RedisDsn.build(
scheme="redis",
host=values.get("REDIS_ADDRESS")
)
# rabbitMQ
RABBITMQ_ADDRESS: str
...
# 短信服务
# 邮件服务
# 支付宝信息
# 存储服务信息
# pip freeze > requirements.txt
# settings = Settings(_env_file='../.env.development', _env_file_encoding='utf-8')
settings = Settings(_env_file='.env.development', _env_file_encoding='utf-8')