使用FastAPI与SQLAlchemy查询现有Oracle数据库表

pip install fastapi uvicorn sqlalchemy cx_Oracle

import cx_Oracle
# 请将此路径替换为您的Oracle Instant Client的实际lib_dir路径
# 例如：Windows系统下 r"C:\oracle\instantclient_19_12"
# Linux/macOS系统下 "/opt/oracle/instantclient_19_12"
cx_Oracle.init_oracle_client(lib_dir=r"E:\instantclient-basic-windows.x64-12.1.0.2.0\instantclient_12_1")

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# 替换为您的Oracle数据库连接信息
# 示例：oracle+cx_oracle://用户名:密码@主机名:端口号/服务名
my_database_connection = "oracle+cx_oracle://super:your_password@localhost:1521/db"

engine = create_engine(my_database_connection)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

from fastapi import Depends, FastAPI
from sqlalchemy import Column, Integer, String

# ... (前文的cx_Oracle初始化、engine和SessionLocal定义) ...

class table1(Base):
    __tablename__ = "table1" # 映射到数据库中名为"table1"的表
    # 定义表的列，注意主键是必须的，这里假设CN是主键或唯一标识
    # 如果没有显式主键，SQLAlchemy可能无法正确操作
    CN = Column(String(length=256), primary_key=True) # 假设CN是主键
    NAME = Column(String(length=40))
    EMAIL = Column(String(length=20))

# 即使table1已存在，此行代码在默认checkfirst=True的情况下也不会报错
# 它会检查表是否存在，如果存在则跳过创建
Base.metadata.create_all(bind=engine)

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/table_api/")
def read_table1_data(db: Session = Depends(get_db)):
    """
    查询table1中的所有数据
    """
    cn_information = db.query(table1).all()
    return cn_information

from sqlalchemy import Table, MetaData
from sqlalchemy.orm import declarative_base

# ... (前文的cx_Oracle初始化、engine和SessionLocal定义) ...

Base = declarative_base()
metadata = MetaData() # 创建一个MetaData实例

# 显式反射现有表table1
# autoload_with=engine 会让SQLAlchemy连接数据库并自动加载表的列信息
class table1_reflected(Base):
    __table__ = Table("table1", metadata, autoload_with=engine)
    # 可以在这里添加额外的ORM属性或关系，但列定义会从数据库中反射

# 注意：使用反射机制时，通常不需要调用Base.metadata.create_all()
# 因为我们是基于已存在的表进行映射，而不是创建新表。
# 如果您在同一个应用中混合使用了创建和反射，请谨慎管理MetaData实例。

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/table_api_reflected/")
def read_table1_reflected_data(db: Session = Depends(get_db)):
    """
    查询通过反射机制映射的table1数据
    """
    # 查询时使用反射后的模型类
    cn_information = db.query(table1_reflected).all()
    return cn_information

# ... (上述SQLAlchemy模型定义) ...

def get_db():
    db = SessionLocal()
    try:
        yield db # 将数据库会话传递给路由函数
    finally:
        db.close() # 请求结束后关闭会话

# ... (上述FastAPI应用实例和get_db函数) ...

@app.get("/table_api/")
def read_table_data(db: Session = Depends(get_db)):
    """
    查询table1中的所有数据并返回
    """
    # 使用SQLAlchemy会话执行查询
    cn_information = db.query(table1).all() # 使用之前定义的table1模型
    return cn_information

uvicorn main:app --reload