使用FastAPI和SQLAlchemy查询Oracle现有数据库表：映射与实践

pip install fastapi uvicorn sqlalchemy cx_Oracle

import cx_Oracle
from sqlalchemy import create_engine, Column, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from fastapi import Depends, FastAPI

# 配置Oracle Instant Client路径
# 请将此路径替换为您的Instant Client实际路径
cx_Oracle.init_oracle_client(lib_dir=r"E:\instantclient-basic-windows.x64-12.1.0.2.0\instantclient_12_1")

# Oracle数据库连接字符串
# 格式: oracle+cx_oracle://user:password@host:port/service_name
# 请替换为您的实际数据库连接信息
DATABASE_CONNECTION_STRING = "oracle+cx_oracle://your_user:your_password@your_host:1521/your_service_name"

# 创建SQLAlchemy引擎
engine = create_engine(DATABASE_CONNECTION_STRING)

# 创建会话本地工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建声明式基类
Base = declarative_base()

# FastAPI 应用实例
app = FastAPI()

# 数据库会话依赖函数
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

from sqlalchemy import Table, MetaData

# ... (前面的环境准备和数据库连接代码保持不变) ...

# 重新定义Base，以确保Metadata是可用的
Base = declarative_base()
metadata = Base.metadata # 获取metadata对象

# 定义一个映射到现有表table1的类
class Table1(Base):
    __tablename__ = "table1"
    # 使用反射机制，从数据库中加载表结构
    # 注意：这里不需要手动定义所有Column，__table__会自动加载
    __table__ = Table(__tablename__, metadata, autoload_with=engine)

    # 如果需要，可以为ORM操作添加额外的方法或属性
    # 例如，定义主键（如果__table__反射时未自动识别）
    # CN = Column(String(256), primary_key=True) # 假设CN是主键

    def __repr__(self):
        # 假设CN和NAME是表中的列
        return f"<Table1(CN='{self.CN}', NAME='{self.NAME}')>"

# 注意：对于反射的表，通常不需要调用Base.metadata.create_all()
# 因为表已经存在，且我们是通过反射来映射它。
# 如果您确实想保留create_all()，如前所述，它会因checkfirst=True而跳过。
# 但为了清晰起见，在纯反射场景下可以省略。
# Base.metadata.create_all(bind=engine) # 这一行在反射场景下可以移除或保持不变

# ... (FastAPI应用和get_db依赖函数保持不变) ...

@app.get("/table1_data/")
def read_table1_data(db: Session = Depends(get_db)):
    """
    查询table1的所有数据
    """
    try:
        # 使用反射映射的Table1类进行查询
        data = db.query(Table1).all()
        # 将查询结果转换为字典列表，以便FastAPI可以序列化
        return [{"CN": item.CN, "NAME": item.NAME, "EMAIL": item.EMAIL} for item in data]
    except Exception as e:
        # 简单的错误处理
        return {"error": str(e)}

# 运行FastAPI应用 (在命令行执行: uvicorn your_module_name:app --reload)

# ... (前面的环境准备和数据库连接代码保持不变) ...

Base = declarative_base()

class Table1(Base):
    __tablename__ = "table1"
    # 手动定义所有列，必须与数据库中的实际列名和类型匹配
    CN = Column(String(length=256), primary_key=True) # 假设CN是主键
    NAME = Column(String(length=40))
    EMAIL = Column(String(length=20))

    def __repr__(self):
        return f"<Table1(CN='{self.CN}', NAME='{self.NAME}')>"

# 在这种情况下，Base.metadata.create_all(bind=engine) 可以保留。
# 它会检查table1是否存在，如果存在则跳过创建。
# 重要的是，您的Python模型定义必须与数据库中的表结构完全一致。
Base.metadata.create_all(bind=engine)

# ... (FastAPI应用和get_db依赖函数保持不变) ...

@app.get("/table1_data_manual/")
def read_table1_data_manual(db: Session = Depends(get_db)):
    """
    查询table1的所有数据 (使用手动定义模型)
    """
    try:
        data = db.query(Table1).all()
        return [{"CN": item.CN, "NAME": item.NAME, "EMAIL": item.EMAIL} for item in data]
    except Exception as e:
        return {"error": str(e)}

import cx_Oracle
from sqlalchemy import create_engine, Column, String, Table, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from fastapi import Depends, FastAPI
from typing import List, Dict, Any

# --- 1. 环境与数据库配置 ---
cx_Oracle.init_oracle_client(lib_dir=r"E:\instantclient-basic-windows.x64-12.1.0.2.0\instantclient_12_1") # 替换为您的路径

# 注意：请务必替换为您的实际数据库连接信息，并考虑使用环境变量或配置文件来管理敏感信息
DATABASE_CONNECTION_STRING = "oracle+cx_oracle://your_user:your_password@your_host:1521/your_service_name"

engine = create_engine(DATABASE_CONNECTION_STRING)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
metadata = Base.metadata # 获取metadata对象

app = FastAPI(title="Oracle数据库查询API")

# --- 2. 数据库会话依赖 ---
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# --- 3. 映射现有Oracle表 (使用反射机制) ---
class Table1(Base):
    __tablename__ = "table1"
    __table__ = Table(__tablename__, metadata, autoload_with=engine)

    def __repr__(self):
        # 假设反射后可以通过属性访问列
        cn_val = getattr(self, 'CN', 'N/A')
        name_val = getattr(self, 'NAME', 'N/A')
        return f"<Table1(CN='{cn_val}', NAME='{name_val}')>"

# --- 4. FastAPI 接口定义 ---
@app.get("/table1_data/", response_model=List[Dict[str, Any]], summary="查询Oracle中table1的所有数据")
def read_table1_data(db: Session = Depends(get_db)):
    """
    从Oracle数据库的`table1`中查询所有记录。
    返回一个包含每行数据字典的列表。
    """
    try:
        # 执行查询
        data = db.query(Table1).all()

        # 将ORM对象转换为字典列表，方便FastAPI序列化和前端消费
        # 遍历__table__.columns获取所有列名
        result_list = []
        for item in data:
            row_dict = {}
            for column in Table1.__table__.columns:
                row_dict[column.name] = getattr(item, column.name)
            result_list.append(row_dict)
        return result_list
    except Exception as e:
        print(f"查询出错: {e}") # 打印错误到控制台
        raise HTTPException(status_code=500, detail=f"数据库查询失败: {e}")

# 如果有其他表，可以类似地定义
class Table2(Base):
    __tablename__ = "table2"
    __table__ = Table(__tablename__, metadata, autoload_with=engine)
    # ... 其他方法或属性 ...

@app.get("/table2_data/", response_model=List[Dict[str, Any]], summary="查询Oracle中table2的所有数据")
def read_table2_data(db: Session = Depends(get_db)):
    """
    从Oracle数据库的`table2`中查询所有记录。
    """
    try:
        data = db.query(Table2).all()
        result_list = []
        for item in data:
            row_dict = {}
            for column in Table2.__table__.columns:
                row_dict[column.name] = getattr(item, column.name)
            result_list.append(row_dict)
        return result_list
    except Exception as e:
        print(f"查询出错: {e}")
        raise HTTPException(status_code=500, detail=f"数据库查询失败: {e}")

# 导入FastAPI的HTTPException
from fastapi import HTTPException