Simplify apps and plugins config method

.pre-commit-config.yaml

@@ -8,7 +8,7 @@ repos:
       - id: check-toml
 
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: v0.11.2
+    rev: v0.11.9
     hooks:
       - id: ruff
         args:
@@ -20,7 +20,7 @@ repos:
       - id: ruff-format
 
   - repo: https://github.com/astral-sh/uv-pre-commit
-    rev: 0.6.10
+    rev: 0.7.3
     hooks:
       - id: uv-lock
       - id: uv-export

backend/.env.example

@@ -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

@@ -5,9 +5,9 @@
 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

@@ -5,14 +5,14 @@
 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

@@ -5,16 +5,16 @@
 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/service/auth_service.py

@@ -5,7 +5,6 @@
 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 @@ async def login(
             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

@@ -5,7 +5,6 @@
 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 @@ async def create_with_login(
                 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

@@ -191,7 +191,7 @@ async def update_avatar(*, request: Request, username: str, avatar: AvatarParam)
         """
         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

@@ -5,39 +5,38 @@
 import celery
 import celery_aio_pool
 
-from backend.app.task.conf import task_settings
 from backend.core.conf import settings
 
 __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}'
     )
 
 
 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

@@ -7,15 +7,15 @@
 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/common/socketio/server.py

@@ -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 @@
     # 集成 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',