public/fastapi-best-architecture
1
0
Fork 0
mirror of https://github.com/fastapi-practices/fastapi-best-architecture.git synced 2026-09-21 21:15:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simplify apps and plugins config method (#600)

Browse Source
This commit is contained in:
Wu Clan
2025-05-13 13:16:06 +08:00
committed by GitHub
parent 87f29e52a1
commit 27b9cf4c64
23 changed files with 318 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.env.example
+6 -6
View File
@@ -15,18 +15,18 @@ REDIS_DATABASE=0
TOKEN_SECRET_KEY='1VkVF75nsNABBjK_7-qz7GtzNy3AMvktc9TCPwKczCk'
# Opera Log
OPERA_LOG_ENCRYPT_SECRET_KEY='d77b25790a804c2b4a339dd0207941e4cefa5751935a33735bc73bb7071a005b'
# Admin
# App Admin
# OAuth2
OAUTH2_GITHUB_CLIENT_ID='test'
OAUTH2_GITHUB_CLIENT_SECRET='test'
OAUTH2_LINUX_DO_CLIENT_ID='test'
OAUTH2_LINUX_DO_CLIENT_SECRET='test'
# Task
# App Task
# Celery
CELERY_BROKER_REDIS_DATABASE=1
CELERY_BACKEND_REDIS_DATABASE=2
# Rabbitmq
RABBITMQ_HOST='127.0.0.1'
RABBITMQ_PORT=5672
RABBITMQ_USERNAME='guest'
RABBITMQ_PASSWORD='guest'
CELERY_RABBITMQ_HOST='127.0.0.1'
CELERY_RABBITMQ_PORT=5672
CELERY_RABBITMQ_USERNAME='guest'
CELERY_RABBITMQ_PASSWORD='guest'
backend/app/admin/api/v1/auth/captcha.py
+3 -3
View File
@@ -5,9 +5,9 @@ from fastapi import APIRouter, Depends, Request
from fastapi_limiter.depends import RateLimiter
from starlette.concurrency import run_in_threadpool
from backend.app.admin.conf import admin_settings
from backend.app.admin.schema.captcha import GetCaptchaDetail
from backend.common.response.response_schema import ResponseSchemaModel, response_base
from backend.core.conf import settings
from backend.database.redis import redis_client
router = APIRouter()
@@ -26,9 +26,9 @@ async def get_captcha(request: Request) -> ResponseSchemaModel[GetCaptchaDetail]
img, code = await run_in_threadpool(img_captcha, img_byte=img_type)
ip = request.state.ip
await redis_client.set(
f'{admin_settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{ip}',
f'{settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{ip}',
code,
ex=admin_settings.CAPTCHA_LOGIN_EXPIRE_SECONDS,
ex=settings.CAPTCHA_LOGIN_EXPIRE_SECONDS,
)
data = GetCaptchaDetail(image_type=img_type, image=img)
return response_base.success(data=data)
backend/app/admin/api/v1/oauth2/github.py
+3 -3
View File
@@ -5,14 +5,14 @@ from fastapi_limiter.depends import RateLimiter
from fastapi_oauth20 import FastAPIOAuth20, GitHubOAuth20
from starlette.responses import RedirectResponse
from backend.app.admin.conf import admin_settings
from backend.app.admin.service.oauth2_service import oauth2_service
from backend.common.enums import UserSocialType
from backend.common.response.response_schema import ResponseSchemaModel, response_base
from backend.core.conf import settings
router = APIRouter()
_github_client = GitHubOAuth20(admin_settings.OAUTH2_GITHUB_CLIENT_ID, admin_settings.OAUTH2_GITHUB_CLIENT_SECRET)
_github_client = GitHubOAuth20(settings.OAUTH2_GITHUB_CLIENT_ID, settings.OAUTH2_GITHUB_CLIENT_SECRET)
_github_oauth2 = FastAPIOAuth20(_github_client, redirect_route_name='github_login')
@@ -44,4 +44,4 @@ async def github_login(
user=user,
social=UserSocialType.github,
)
return RedirectResponse(url=f'{admin_settings.OAUTH2_FRONTEND_REDIRECT_URI}?access_token={data.access_token}')
return RedirectResponse(url=f'{settings.OAUTH2_FRONTEND_REDIRECT_URI}?access_token={data.access_token}')
backend/app/admin/api/v1/oauth2/linux_do.py
+4 -4
View File
@@ -5,16 +5,16 @@ from fastapi_limiter.depends import RateLimiter
from fastapi_oauth20 import FastAPIOAuth20, LinuxDoOAuth20
from starlette.responses import RedirectResponse
from backend.app.admin.conf import admin_settings
from backend.app.admin.service.oauth2_service import oauth2_service
from backend.common.enums import UserSocialType
from backend.common.response.response_schema import ResponseSchemaModel, response_base
from backend.core.conf import settings
router = APIRouter()
_linux_do_client = LinuxDoOAuth20(
admin_settings.OAUTH2_LINUX_DO_CLIENT_ID,
admin_settings.OAUTH2_LINUX_DO_CLIENT_SECRET,
settings.OAUTH2_LINUX_DO_CLIENT_ID,
settings.OAUTH2_LINUX_DO_CLIENT_SECRET,
)
_linux_do_oauth2 = FastAPIOAuth20(_linux_do_client, redirect_route_name='linux_do_login')
@@ -47,4 +47,4 @@ async def linux_do_login(
user=user,
social=UserSocialType.linux_do,
)
return RedirectResponse(url=f'{admin_settings.OAUTH2_FRONTEND_REDIRECT_URI}?access_token={data.access_token}')
return RedirectResponse(url=f'{settings.OAUTH2_FRONTEND_REDIRECT_URI}?access_token={data.access_token}')
backend/app/admin/conf.py
-35
View File
@@ -1,35 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict
from backend.core.path_conf import BASE_PATH
class AdminSettings(BaseSettings):
"""Admin 配置"""
model_config = SettingsConfigDict(env_file=f'{BASE_PATH}/.env', env_file_encoding='utf-8', extra='ignore')
# .env OAuth2
OAUTH2_GITHUB_CLIENT_ID: str
OAUTH2_GITHUB_CLIENT_SECRET: str
OAUTH2_LINUX_DO_CLIENT_ID: str
OAUTH2_LINUX_DO_CLIENT_SECRET: str
# OAuth2
OAUTH2_FRONTEND_REDIRECT_URI: str = 'http://localhost:5173/oauth2/callback'
# 验证码
CAPTCHA_LOGIN_REDIS_PREFIX: str = 'fba:login:captcha'
CAPTCHA_LOGIN_EXPIRE_SECONDS: int = 60 * 5 # 3 分钟
@lru_cache
def get_admin_settings() -> AdminSettings:
"""获取 admin 配置"""
return AdminSettings()
admin_settings = get_admin_settings()
backend/app/admin/service/auth_service.py
+2 -3
View File
@@ -5,7 +5,6 @@ from fastapi.security import HTTPBasicCredentials
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.background import BackgroundTask, BackgroundTasks
from backend.app.admin.conf import admin_settings
from backend.app.admin.crud.crud_user import user_dao
from backend.app.admin.model import User
from backend.app.admin.schema.token import GetLoginToken, GetNewToken
@@ -91,12 +90,12 @@ class AuthService:
user = None
try:
user = await self.user_verify(db, obj.username, obj.password)
captcha_code = await redis_client.get(f'{admin_settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{request.state.ip}')
captcha_code = await redis_client.get(f'{settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{request.state.ip}')
if not captcha_code:
raise errors.AuthorizationError(msg='验证码失效，请重新获取')
if captcha_code.lower() != obj.captcha.lower():
raise errors.CustomError(error=CustomErrorCode.CAPTCHA_ERROR)
await redis_client.delete(f'{admin_settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{request.state.ip}')
await redis_client.delete(f'{settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{request.state.ip}')
await user_dao.update_login_time(db, obj.username)
await db.refresh(user)
a_token = await create_access_token(
backend/app/admin/service/oauth2_service.py
+1 -2
View File
@@ -5,7 +5,6 @@ from typing import Any
from fast_captcha import text_captcha
from fastapi import BackgroundTasks, Request, Response
from backend.app.admin.conf import admin_settings
from backend.app.admin.crud.crud_user import user_dao
from backend.app.admin.crud.crud_user_social import user_social_dao
from backend.app.admin.schema.token import GetLoginToken
@@ -102,7 +101,7 @@ class OAuth2Service:
msg='登录成功（OAuth2）',
)
background_tasks.add_task(login_log_service.create, **login_log)
await redis_client.delete(f'{admin_settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{request.state.ip}')
await redis_client.delete(f'{settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{request.state.ip}')
response.set_cookie(
key=settings.COOKIE_REFRESH_TOKEN_KEY,
value=refresh_token.refresh_token,
backend/app/admin/service/user_service.py
+1 -1
View File
@@ -191,7 +191,7 @@ class UserService:
"""
async with async_db_session.begin() as db:
if request.user.username != username:
raise errors.AuthorizationError
raise errors.AuthorizationError(msg='你只能修改自己的信息')
user = await user_dao.get_by_username(db, username)
if not user:
raise errors.NotFoundError(msg='用户不存在')
backend/app/task/celery.py
+9 -10
View File
@@ -5,7 +5,6 @@ from typing import Any
import celery
import celery_aio_pool
from backend.app.task.conf import task_settings
from backend.core.conf import settings
__all__ = ['celery_app']
@@ -13,14 +12,14 @@ __all__ = ['celery_app']
def get_broker_url() -> str:
"""获取消息代理 URL"""
if task_settings.CELERY_BROKER == 'redis':
if settings.CELERY_BROKER == 'redis':
return (
f'redis://:{settings.REDIS_PASSWORD}@{settings.REDIS_HOST}:'
f'{settings.REDIS_PORT}/{task_settings.CELERY_BROKER_REDIS_DATABASE}'
f'{settings.REDIS_PORT}/{settings.CELERY_BROKER_REDIS_DATABASE}'
)
return (
f'amqp://{task_settings.RABBITMQ_USERNAME}:{task_settings.RABBITMQ_PASSWORD}@'
f'{task_settings.RABBITMQ_HOST}:{task_settings.RABBITMQ_PORT}'
f'amqp://{settings.CELERY_RABBITMQ_USERNAME}:{settings.CELERY_RABBITMQ_PASSWORD}@'
f'{settings.CELERY_RABBITMQ_HOST}:{settings.CELERY_RABBITMQ_PORT}'
)
@@ -28,16 +27,16 @@ def get_result_backend() -> str:
"""获取结果后端 URL"""
return (
f'redis://:{settings.REDIS_PASSWORD}@{settings.REDIS_HOST}:'
f'{settings.REDIS_PORT}/{task_settings.CELERY_BACKEND_REDIS_DATABASE}'
f'{settings.REDIS_PORT}/{settings.CELERY_BACKEND_REDIS_DATABASE}'
)
def get_result_backend_transport_options() -> dict[str, Any]:
"""获取结果后端传输选项"""
return {
'global_keyprefix': task_settings.CELERY_BACKEND_REDIS_PREFIX,
'global_keyprefix': settings.CELERY_BACKEND_REDIS_PREFIX,
'retry_policy': {
'timeout': task_settings.CELERY_BACKEND_REDIS_TIMEOUT,
'timeout': settings.CELERY_BACKEND_REDIS_TIMEOUT,
},
}
@@ -55,7 +54,7 @@ def init_celery() -> celery.Celery:
'fba_celery',
enable_utc=False,
timezone=settings.DATETIME_TIMEZONE,
beat_schedule=task_settings.CELERY_SCHEDULE,
beat_schedule=settings.CELERY_SCHEDULE,
broker_url=get_broker_url(),
broker_connection_retry_on_startup=True,
result_backend=get_result_backend(),
@@ -65,7 +64,7 @@ def init_celery() -> celery.Celery:
)
# 自动发现任务
app.autodiscover_tasks(task_settings.CELERY_TASK_PACKAGES)
app.autodiscover_tasks(settings.CELERY_TASK_PACKAGES)
return app
backend/app/task/celery_task/base.py
+2 -2
View File
@@ -7,15 +7,15 @@ from typing import Any
from celery import Task
from sqlalchemy.exc import SQLAlchemyError
from backend.app.task.conf import task_settings
from backend.common.socketio.actions import task_notification
from backend.core.conf import settings
class TaskBase(Task):
"""Celery 任务基类"""
autoretry_for = (SQLAlchemyError,)
max_retries = task_settings.CELERY_TASK_MAX_RETRIES
max_retries = settings.CELERY_TASK_MAX_RETRIES
async def before_start(self, task_id: str, args, kwargs) -> None:
"""
backend/app/task/conf.py
-73
View File
@@ -1,73 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from functools import lru_cache
from typing import Any, Literal
from celery.schedules import crontab
from pydantic import model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
from backend.core.path_conf import BASE_PATH
class TaskSettings(BaseSettings):
"""Celery 任务配置"""
model_config = SettingsConfigDict(env_file=f'{BASE_PATH}/.env', env_file_encoding='utf-8', extra='ignore')
# .env 环境
ENVIRONMENT: Literal['dev', 'pro']
# .env Redis 配置
CELERY_BROKER_REDIS_DATABASE: int
CELERY_BACKEND_REDIS_DATABASE: int
# .env RabbitMQ 配置
# docker run -d --hostname fba-mq --name fba-mq -p 5672:5672 -p 15672:15672 rabbitmq:latest
RABBITMQ_HOST: str
RABBITMQ_PORT: int
RABBITMQ_USERNAME: str
RABBITMQ_PASSWORD: str
# Celery 基础配置
CELERY_BROKER: Literal['rabbitmq', 'redis'] = 'redis'
CELERY_BACKEND_REDIS_PREFIX: str = 'fba:celery:'
CELERY_BACKEND_REDIS_TIMEOUT: int = 5
CELERY_TASK_PACKAGES: list[str] = [
'app.task.celery_task',
'app.task.celery_task.db_log',
]
CELERY_TASK_MAX_RETRIES: int = 5
# Celery 定时任务配置
CELERY_SCHEDULE: dict[str, dict[str, Any]] = {
'exec-every-10-seconds': {
'task': 'task_demo_async',
'schedule': 10,
},
'exec-every-sunday': {
'task': 'delete_db_opera_log',
'schedule': crontab('0', '0', day_of_week='6'),
},
'exec-every-15-of-month': {
'task': 'delete_db_login_log',
'schedule': crontab('0', '0', day_of_month='15'),
},
}
@model_validator(mode='before')
@classmethod
def validate_celery_broker(cls, values: Any) -> Any:
"""生产环境强制使用 RabbitMQ 作为消息代理"""
if values['ENVIRONMENT'] == 'pro':
values['CELERY_BROKER'] = 'rabbitmq'
return values
@lru_cache
def get_task_settings() -> TaskSettings:
"""获取 Celery 任务配置"""
return TaskSettings()
task_settings = get_task_settings()
backend/common/socketio/server.py
+4 -5
View File
@@ -2,7 +2,6 @@
# -*- coding: utf-8 -*-
import socketio
from backend.app.task.conf import task_settings
from backend.common.log import log
from backend.common.security.jwt import jwt_authentication
from backend.core.conf import settings
@@ -13,13 +12,13 @@ sio = socketio.AsyncServer(
# 集成 Celery 实现消息订阅
client_manager=socketio.AsyncRedisManager(
f'redis://:{settings.REDIS_PASSWORD}@{settings.REDIS_HOST}:'
f'{settings.REDIS_PORT}/{task_settings.CELERY_BROKER_REDIS_DATABASE}'
f'{settings.REDIS_PORT}/{settings.CELERY_BROKER_REDIS_DATABASE}'
)
if task_settings.CELERY_BROKER == 'redis'
if settings.CELERY_BROKER == 'redis'
else socketio.AsyncAioPikaManager(
(
f'amqp://{task_settings.RABBITMQ_USERNAME}:{task_settings.RABBITMQ_PASSWORD}@'
f'{task_settings.RABBITMQ_HOST}:{task_settings.RABBITMQ_PORT}'
f'amqp://{settings.CELERY_RABBITMQ_USERNAME}:{settings.CELERY_RABBITMQ_PASSWORD}@'
f'{settings.CELERY_RABBITMQ_HOST}:{settings.CELERY_RABBITMQ_PORT}'
)
),
async_mode='asgi',
backend/core/conf.py
+66 -1
View File
@@ -3,6 +3,7 @@
from functools import lru_cache
from typing import Any, Literal
from celery.schedules import crontab
from pydantic import model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -184,13 +185,77 @@ class Settings(BaseSettings):
PLUGIN_PIP_CHINA: bool = True
PLUGIN_PIP_INDEX_URL: str = 'https://mirrors.aliyun.com/pypi/simple/'
# App Admin
# .env OAuth2
OAUTH2_GITHUB_CLIENT_ID: str
OAUTH2_GITHUB_CLIENT_SECRET: str
OAUTH2_LINUX_DO_CLIENT_ID: str
OAUTH2_LINUX_DO_CLIENT_SECRET: str
# OAuth2
OAUTH2_FRONTEND_REDIRECT_URI: str = 'http://localhost:5173/oauth2/callback'
# 验证码
CAPTCHA_LOGIN_REDIS_PREFIX: str = 'fba:login:captcha'
CAPTCHA_LOGIN_EXPIRE_SECONDS: int = 60 * 5 # 3 分钟
# App Task
# .env Redis 配置
CELERY_BROKER_REDIS_DATABASE: int
CELERY_BACKEND_REDIS_DATABASE: int
# .env RabbitMQ 配置
# docker run -d --hostname fba-mq --name fba-mq -p 5672:5672 -p 15672:15672 rabbitmq:latest
CELERY_RABBITMQ_HOST: str
CELERY_RABBITMQ_PORT: int
CELERY_RABBITMQ_USERNAME: str
CELERY_RABBITMQ_PASSWORD: str
# 基础配置
CELERY_BROKER: Literal['rabbitmq', 'redis'] = 'redis'
CELERY_BACKEND_REDIS_PREFIX: str = 'fba:celery:'
CELERY_BACKEND_REDIS_TIMEOUT: int = 5
CELERY_TASK_PACKAGES: list[str] = [
'app.task.celery_task',
'app.task.celery_task.db_log',
]
CELERY_TASK_MAX_RETRIES: int = 5
# 定时任务配置
CELERY_SCHEDULE: dict[str, dict[str, Any]] = {
'exec-every-10-seconds': {
'task': 'task_demo_async',
'schedule': 10,
},
'exec-every-sunday': {
'task': 'delete_db_opera_log',
'schedule': crontab('0', '0', day_of_week='6'),
},
'exec-every-15-of-month': {
'task': 'delete_db_login_log',
'schedule': crontab('0', '0', day_of_month='15'),
},
}
# Plugin Code Generator
# 代码下载
CODE_GENERATOR_DOWNLOAD_ZIP_FILENAME: str = 'fba_generator'
# Plugin Config
# 参数配置
CONFIG_BUILT_IN_TYPES: list[str] = ['website', 'protocol', 'policy']
@model_validator(mode='before')
@classmethod
def check_env(cls, values: Any) -> Any:
"""生产环境下禁用 OpenAPI 文档和静态文件服务"""
"""检查环境变量"""
if values.get('ENVIRONMENT') == 'pro':
# FastAPI
values['FASTAPI_OPENAPI_URL'] = None
values['FASTAPI_STATIC_FILES'] = False
# Task
values['CELERY_BROKER'] = 'rabbitmq'
return values
backend/plugin/code_generator/api/v1/gen.py
+2 -2
View File
@@ -9,7 +9,7 @@ from backend.common.response.response_schema import ResponseModel, ResponseSchem
from backend.common.security.jwt import DependsJwtAuth
from backend.common.security.permission import RequestPermission
from backend.common.security.rbac import DependsRBAC
from backend.plugin.code_generator.conf import generator_settings
from backend.core.conf import settings
from backend.plugin.code_generator.schema.gen import ImportParam
from backend.plugin.code_generator.service.gen_service import gen_service
@@ -69,5 +69,5 @@ async def download_code(pk: Annotated[int, Path(description='业务 ID')]):
return StreamingResponse(
bio,
media_type='application/x-zip-compressed',
headers={'Content-Disposition': f'attachment; filename={generator_settings.DOWNLOAD_ZIP_FILENAME}.zip'},
headers={'Content-Disposition': f'attachment; filename={settings.CODE_GENERATOR_DOWNLOAD_ZIP_FILENAME}.zip'},
)
backend/plugin/code_generator/conf.py
-21
View File
@@ -1,21 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from functools import lru_cache
from pydantic_settings import BaseSettings
class GeneratorSettings(BaseSettings):
"""代码生成配置"""
# 代码下载
DOWNLOAD_ZIP_FILENAME: str = 'fba_generator'
@lru_cache
def get_generator_settings() -> GeneratorSettings:
"""获取代码生成配置"""
return GeneratorSettings()
generator_settings = get_generator_settings()
backend/plugin/config/conf.py
-21
View File
@@ -1,21 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from functools import lru_cache
from pydantic_settings import BaseSettings
class ConfigSettings(BaseSettings):
"""参数配置"""
# 参数
CONFIG_BUILT_IN_TYPES: list[str] = ['website', 'protocol', 'policy']
@lru_cache
def get_config_settings() -> ConfigSettings:
"""获取参数配置"""
return ConfigSettings()
config_settings = get_config_settings()
backend/plugin/config/crud/crud_config.py
+4 -4
View File
@@ -6,7 +6,7 @@ from sqlalchemy import Select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy_crud_plus import CRUDPlus
from backend.plugin.config.conf import config_settings
from backend.core.conf import settings
from backend.plugin.config.model import Config
from backend.plugin.config.schema.config import CreateConfigParam, UpdateConfigParam
@@ -22,7 +22,7 @@ class CRUDConfig(CRUDPlus[Config]):
:param pk: 参数配置 ID
:return:
"""
return await self.select_model_by_column(db, id=pk, type__not_in=config_settings.CONFIG_BUILT_IN_TYPES)
return await self.select_model_by_column(db, id=pk, type__not_in=settings.CONFIG_BUILT_IN_TYPES)
async def get_by_type(self, db: AsyncSession, type: str) -> Sequence[Config]:
"""
@@ -63,7 +63,7 @@ class CRUDConfig(CRUDPlus[Config]):
:param type: 参数配置类型
:return:
"""
filters = {'type__not_in': config_settings.CONFIG_BUILT_IN_TYPES}
filters = {'type__not_in': settings.CONFIG_BUILT_IN_TYPES}
if name is not None:
filters.update(name__like=f'%{name}%')
if type is not None:
@@ -100,7 +100,7 @@ class CRUDConfig(CRUDPlus[Config]):
:return:
"""
return await self.delete_model_by_column(
db, allow_multiple=True, id__in=pk, type__not_in=config_settings.CONFIG_BUILT_IN_TYPES
db, allow_multiple=True, id__in=pk, type__not_in=settings.CONFIG_BUILT_IN_TYPES
)
backend/plugin/config/service/config_service.py
+2 -2
View File
@@ -5,8 +5,8 @@ from typing import Sequence
from sqlalchemy import Select
from backend.common.exception import errors
from backend.core.conf import settings
from backend.database.db import async_db_session
from backend.plugin.config.conf import config_settings
from backend.plugin.config.crud.crud_config import config_dao
from backend.plugin.config.model import Config
from backend.plugin.config.schema.config import (
@@ -83,7 +83,7 @@ class ConfigService:
:return:
"""
async with async_db_session.begin() as db:
if obj.type in config_settings.CONFIG_BUILT_IN_TYPES:
if obj.type in settings.CONFIG_BUILT_IN_TYPES:
raise errors.ForbiddenError(msg='非法类型参数')
config = await config_dao.get_by_key(db, obj.key)
if config:
backend/scripts/init_data.py
-1
View File
@@ -1,6 +1,5 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ruff: noqa: I001
from anyio import run
from backend.database.db import create_table
backend/scripts/init_plugin.py
-1
View File
@@ -1,6 +1,5 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ruff: noqa: I001
from anyio import run
from backend.plugin.tools import install_requirements_async