update：后端-更新config配置模式

config/development.yaml

@@ -0,0 +1,44 @@
+# 通用配置
+common:
+  env: development
+  log_level: DEBUG
+
+# FastAPI 服务器配置
+server:
+  host: 0.0.0.0
+  port: 8000
+  debug: true
+  # token信息
+  secret_key: "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
+  algorithm: "HS256"
+  access_token_expire_minutes: 30
+  # 连接数据库
+  database_uri: "mysql+pymysql://root:123456@localhost/devops"
+  casbin_model_path: "server/model.conf"
+  #  白名单
+  no_verify_url:
+    - "/"
+    - "/api/login"
+  redis:
+    host: "192.168.137.129"
+    password: "seraphim"
+    port: 6379
+    health_check_interval: 30
+  #  配置连接rpyc信息
+  rpyc_config:
+    host: localhost
+    port: 18861
+    config:
+      allow_public_attrs: true
+      allow_pickle: true
+    keepalive: true
+
+# RPyC Scheduler 配置
+scheduler:
+  rpc_port: 18861
+  apscheduler_job_store: 'mysql+pymysql://root:[email protected]/devops'
+  redis:
+    host: "192.168.137.129"
+    password: "seraphim"
+    port: 6379
+    health_check_interval: 30

rpyc_scheduler/scheduler-server.py → rpyc_scheduler/server.py

@@ -6,17 +6,13 @@
 
 import rpyc
 from typing import List
+from rpyc.utils.server import ThreadedServer
+from loguru import logger
 
 from apscheduler.triggers.cron import CronTrigger
 from apscheduler.triggers.date import DateTrigger
-from sqlmodel import text
-from tasks import *
-from datetime import datetime
-from loguru import logger
-from config import rpc_config
-from rpyc.utils.server import ThreadedServer
+from rpyc_scheduler.tasks import *
 from apscheduler.job import Job
-# from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
 from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
 from apscheduler.events import (
     EVENT_JOB_EXECUTED,
@@ -25,9 +21,12 @@
     EVENT_JOB_SUBMITTED,
     EVENT_JOB_REMOVED
 )
-from jobstore import CustomJobStore
-from scheduler import CustomBackgroundScheduler
-from models import engine
+from rpyc_scheduler.jobstore import CustomJobStore
+from rpyc_scheduler.scheduler import CustomBackgroundScheduler
+from rpyc_scheduler.config import rpc_config
+
+# 全局变量定义
+scheduler = None
 
 
 def print_text(*args, **kwargs):
@@ -59,10 +58,10 @@ def exposed_add_job(self, func, **kwargs):
             trigger = CronTrigger(minute=values[0], hour=values[1], day=values[2], month=values[3],
                                   day_of_week=values[4], start_date=trigger_args['start_date'],
                                   end_date=trigger_args['end_date'])
-            return scheduler.add_job(func, CronTrigger.from_crontab(trigger), **kwargs)
+            return scheduler.add_job(func, trigger=trigger, **kwargs)
         elif trigger == 'date':
             logger.debug(kwargs)
-            logger.debug(trigger,trigger_args)
+            logger.debug(trigger, trigger_args)
             return scheduler.add_job(func, DateTrigger(
                 run_date=trigger_args['run_date'] if trigger_args is not None else None), **kwargs)
 
@@ -132,9 +131,10 @@ def event_listener(event):
         logger.debug('event error')
 
 
-if __name__ == '__main__':
+def main():
+    global scheduler
     job_store = {
-        'default': CustomJobStore(url=str(rpc_config.apscheduler_job_store))
+        'default': CustomJobStore(url=str(rpc_config['apscheduler_job_store']))
     }
     apscheduler_excutors = {
         'default': ThreadPoolExecutor(20),
@@ -147,11 +147,15 @@ def event_listener(event):
     scheduler.start()
 
     # 启动rpyc服务
-    server = ThreadedServer(SchedulerService, port=rpc_config.rpc_port,
+    server = ThreadedServer(SchedulerService, port=rpc_config['rpc_port'],
                             protocol_config={'allow_public_attrs': True, 'allow_pickle': True})
     try:
         server.start()
     except (KeyboardInterrupt, SystemExit):
         pass
     finally:
         scheduler.shutdown()
+
+
+if __name__ == "__main__":
+    main()

server/common/database.py

@@ -19,7 +19,7 @@ def get_session():
 
 
 def get_rpyc():
-    with rpyc.connect(**settings.rpyc_config) as conn:
+    with rpyc.connect(**settings['rpyc_config']) as conn:
         yield conn
 
 
@@ -42,7 +42,7 @@ def get_or_create(session: Session, model, **kwargs):
 
 
 async def get_redis():
-    redis_conn = redis.Redis(**settings.redis_config)
+    redis_conn = redis.Redis(**settings['redis'])
     try:
         yield redis_conn
     finally:

server/common/security.py

@@ -31,7 +31,7 @@ def auth_check(request: Request = None, ws: WebSocket = None):
     if ws:
         return None
     logger.info(f'request url:{request.url} method:{request.method}')
-    for url in settings.NO_VERIFY_URL:
+    for url in settings['no_verify_url']:
         if url == request.url.path.lower():
             logger.debug(f"{request.url.path} 在白名单中，不需要权限验证")
             return True
@@ -41,7 +41,7 @@ def auth_check(request: Request = None, ws: WebSocket = None):
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authenticated")
 
     try:
-        playload = jwt.decode(param, settings.SECRET_KEY, settings.ALGORITHM)
+        playload = jwt.decode(param, settings['secret_key'], settings['algorithm'])
     except jwt.ExpiredSignatureError as e:
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
     except JWTError as e:
@@ -58,12 +58,12 @@ def create_access_token(data):
     :param data:
     :return:
     """
-    expires_delta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    expires_delta = timedelta(minutes=settings['access_token_expire_minutes'])
     to_encode = data.copy()
     if expires_delta:
         expire = datetime.utcnow() + expires_delta
     else:
         expire = datetime.utcnow() + timedelta(minutes=15)
     to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+    encoded_jwt = jwt.encode(to_encode, settings['secret_key'], algorithm=settings['algorithm'])
     return encoded_jwt

server/settings.py

@@ -1,34 +1,25 @@
+import yaml
+import os
+from pathlib import Path
 import casbin_sqlalchemy_adapter
 import casbin
-import rpyc
-from typing import List
-from pathlib import Path
-from pydantic import MySQLDsn
-from pydantic_settings import BaseSettings
 from sqlmodel import create_engine
 
+# 获取环境变量，默认为 development
+env = os.getenv('ENV', 'development')
+config_path = Path(__file__).parent.parent / 'config' / f'{env}.yaml'
 
-class APISettings(BaseSettings):
-    # token加密相关参数
-    SECRET_KEY: str = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
-    ALGORITHM: str = "HS256"
-    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
-
-    CASBIN_MODEL_PATH: str = "server/model.conf"
-    # sql数据库信息
-    DATABASE_URI: MySQLDsn = "mysql+pymysql://root:123456@localhost/devops"
-    # 白名单，不需要进行任何验证即可访问
-    NO_VERIFY_URL: List = [
-        '/',
-        '/api/login',
-    ]
-    rpyc_config: dict = {'host': 'localhost', 'port': 18861,
-                         'config': {"allow_public_attrs": True, 'allow_pickle': True},
-                         'keepalive': True}
-
-
-settings = APISettings()
+# 读取配置文件
+with open(config_path, 'r', encoding='utf-8') as f:
+    config = yaml.safe_load(f)
+    settings = config['server']
 
-engine = create_engine(str(settings.DATABASE_URI), pool_size=5, max_overflow=10, pool_timeout=30, pool_pre_ping=True)
+engine = create_engine(
+    str(settings['database_uri']), 
+    pool_size=5, 
+    max_overflow=10, 
+    pool_timeout=30, 
+    pool_pre_ping=True
+)
 adapter = casbin_sqlalchemy_adapter.Adapter(engine)
-casbin_enforcer = casbin.Enforcer(settings.CASBIN_MODEL_PATH, adapter)
+casbin_enforcer = casbin.Enforcer(settings['casbin_model_path'], adapter)