python 读取env
1建立env 其中格式用等号 db_host127.0.0.1 db_userroot db_password123456 db_port3306 2安装扩展 pip install python-dotenv 3引入扩展 from dotenv import load_dotenv import os 4加载文件 load_dotenv() 5读取内容 host os.getenv(db_host)from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from dotenv import load_dotenv import os load_dotenv() # 1. 从环境变量读取配置 DB_USER os.getenv(db_user) DB_PASSWORD os.getenv(db_password) DB_HOST os.getenv(db_host) DB_PORT os.getenv(db_port) DB_NAME os.getenv(db_database) #2拼接 数据库连接 SQLALCHEMY_DATABASE_URL fmysqlpymysql://{DB_USER}:{DB_PASSWORD}{DB_HOST}:{DB_PORT}/{DB_NAME} #3 创建数据库引擎 engine create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping True) #4 创建回话工厂 用来操作数据库 SessionLocal sessionmaker(autocommitFalse, autoflushFalse,bindengine) #5 创建声明式基类 Base declarative_base() #6 依赖注入 获取数据库回话 用于路由接口 def get_db(): db SessionLocal() try: yield db finally: db.close()#routers/__init__.py from fastapi import APIRouter from .user import router as user_router # from .product import router as product_router # from .order import router as order_router routers [ user_router, # product_router, # order_router ]#routers/user.py from fastapi import APIRouter, Depends #1创建路由实例 router APIRouter(prefix/users,tags[用户管理]) # 测试路由 router.get(/get_user) def get_user(): print(f这是测试路由)首页 from fastapi import FastAPI from fastapi import FastAPI, Depends from sqlalchemy.orm import Session from database import get_db from models import User from dotenv import load_dotenv import os from routers import routers #加载env 文件 load_dotenv() #读取 host os.getenv(db_host) app FastAPI() for router in routers: app.include_router(router) app.get(/) def read_root(db: Session Depends(get_db)): print(f这是地址吗{host}) users db.query(User).all() return users # 直接返回 ORM 对象列表 return {message: Hello, FastAPI} app.get(/items/{item_id}) def read_item(item_id: int, q: str None): return {item_id: item_id, q: q}

相关新闻