使用FastAPI和SQLAlchemy查询现有Oracle数据库表的最佳实践

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

from fastapi import Depends, FastAPI

# 初始化Oracle客户端，请根据实际路径修改
cx_Oracle.init_oracle_client(lib_dir=r"E:\instantclient-basic-windows.x64-12.1.0.2.0\instantclient_12_1")

# 数据库连接字符串，注意保护敏感信息
# 格式: oracle+cx_oracle://user:password@host:port/service_name
DATABASE_URL = "oracle+cx_oracle://super:your_password@localhost:1521/your_db_service_name"

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

# 定义ORM模型，此处以一个名为'table1'的表为例
class Table1(Base):
    __tablename__ = "table1"
    # 假设'CN'是主键，或者至少是唯一标识符
    CN = Column(String(length=256), primary_key=True) # 假设CN是主键
    NAME = Column(String(length=40))
    EMAIL = Column(String(length=20))

# 这一行是讨论的重点，后续会详细解释
# Base.metadata.create_all(bind=engine)

app = FastAPI()

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

@app.get("/table_api/{cn_value}")
def read_table_data(cn_value: str, db: Session = Depends(get_db)):
    """
    根据CN值查询table1中的数据。
    """
    data = db.query(Table1).filter(Table1.CN == cn_value).first()
    if data is None:
        return {"message": f"No data found for CN: {cn_value}"}
    return data

# 示例：即使table1已存在，此行通常也是安全的
# Base.metadata.create_all(bind=engine)

from sqlalchemy import Table, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session, Mapped, mapped_column
from sqlalchemy import String # 确保导入所需的类型

# ... (其他导入和数据库连接配置保持不变) ...

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

# 声明式映射，结合反射
class Table1(Base):
    __tablename__ = "table1"
    __table__ = Table(__tablename__, metadata, autoload_with=engine)
    # 当使用反射时，通常不需要在此处重复定义Column，SQLAlchemy会从数据库中加载
    # 但如果需要覆盖或添加属性（如primary_key），可以显式定义
    # 例如，如果CN不是主键，但你想在ORM层面指定它为主键：
    # CN: Mapped[str] = mapped_column(String(256), primary_key=True)

    # 也可以在不定义__table__的情况下，通过__table_args__指定反射
    # __table_args__ = {'autoload_with': engine}
    # CN = Column(String(length=256), primary_key=True)
    # NAME = Column(String(length=40))
    # EMAIL = Column(String(length=20))

    # 如果完全依赖反射，可以不定义任何Column，但通常为了类型提示和IDE支持，会定义
    # 假设我们只定义了必要的字段，让反射填充其余的
    CN: Mapped[str] = mapped_column(String(256), primary_key=True) # 假设CN是主键

# 注意：使用反射时，通常不再需要 Base.metadata.create_all(bind=engine)
# 因为我们不是要创建表，而是要映射已存在的表。

app = FastAPI()

# ... (get_db 和 API 路由保持不变) ...

from sqlalchemy import Table, MetaData

metadata = MetaData()
# 反射一个名为 'table1' 的表
reflected_table1 = Table('table1', metadata, autoload_with=engine)

# 现在 reflected_table1 包含了 'table1' 的所有列信息
print(reflected_table1.columns.keys())

from sqlalchemy import create_engine, Column, String, Table, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session, Mapped, mapped_column
import cx_Oracle

from fastapi import Depends, FastAPI, HTTPException

# 初始化Oracle客户端
cx_Oracle.init_oracle_client(lib_dir=r"E:\instantclient-basic-windows.x64-12.1.0.2.0\instantclient_12_1")

# 数据库连接字符串
DATABASE_URL = "oracle+cx_oracle://super:your_password@localhost:1521/your_db_service_name"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
metadata = MetaData() # 创建MetaData实例用于反射

# 定义ORM模型，使用反射映射现有表
class Table1(Base):
    __tablename__ = "table1"
    __table__ = Table(__tablename__, metadata, autoload_with=engine)
    # 如果需要，可以在这里为反射的列添加类型提示，例如：
    CN: Mapped[str] = mapped_column(String(256), primary_key=True, init=False) # init=False 避免在__init__中重复设置
    NAME: Mapped[str] = mapped_column(String(40), init=False)
    EMAIL: Mapped[str] = mapped_column(String(20), init=False)

app = FastAPI()

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

@app.get("/table_api/{cn_value}")
def read_table_data(cn_value: str, db: Session = Depends(get_db)):
    """
    根据CN值查询table1中的数据。
    """
    try:
        data = db.query(Table1).filter(Table1.CN == cn_value).first()
        if data is None:
            raise HTTPException(status_code=404, detail=f"No data found for CN: {cn_value}")
        return data
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Database query error: {str(e)}")

# 运行FastAPI应用: uvicorn your_file_name:app --reload