from fastapi import FastAPI, Depends, HTTPException, status import logging from .database.database import engine from .addons.base.models.ir_module_module import Base from sqlalchemy.orm import Session # 在应用启动时创建表 Base.metadata.create_all(bind=engine) app = FastAPI() #临时测试导入 from server.database.database import check_database_connection from server.addons.base.models.res_config_settings import read_conf from .database.database import get_db from importlib import import_module import os from pathlib import Path import os import importlib.util def init_server_conf(): logging.basicConfig(level=logging.DEBUG, format='%(message)s') logging.info(f"fastapi执行完毕") @app.get("/") async def root(): read_conf() check_database_connection() init_server_conf() # app/main.py import pkgutil import importlib from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base from pathlib import Path # 数据库配置 DATABASE_URL = "sqlite:///./test.db" engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() # 自动加载模型 def load_models(directory): package_name = directory.name for _, module_name, _ in pkgutil.iter_modules([str(directory)]): full_module_name = f"{package_name}.{module_name}" importlib.import_module(full_module_name) # 加载所有模型 models_directory = Path(__file__).parent / "models" load_models(models_directory) # 创建数据库表 Base.metadata.create_all(bind=engine) from fastapi import FastAPI import pkgutil import importlib from pathlib import Path # 自动加载模块 def load_modules(directory): package_name = directory.name for _, module_name, _ in pkgutil.iter_modules([str(directory)]): full_module_name = f"{package_name}.{module_name}" importlib.import_module(full_module_name) # 加载所有模块 modules_directory = Path(__file__).parent / "modules" load_modules(modules_directory) # 添加路由 for _, module_name, _ in pkgutil.iter_modules([str(modules_directory)]): module = importlib.import_module(f"app.modules.{module_name}.routes") if hasattr(module, "router"): app.include_router(module.router)