Python中如何在Flask项目中使用SQLAlchemy实现Oracle多用户访问？
一般对于一个 orm 框架的项目来说，模型代码
class User(db.Model):
__tablename__ = 'user'
id = Column(BigInteger, primary_key=True)
a = Column(String)
最终程序会发给数据库 SQL 代码
SELECT user.id AS user_id, user.a AS user_a
FROM user
但是在 oracle 中，user 表前面存在一个数据库的用户概念（比如说 user1 ），所以，必须发送如下代码才能获取数据
SELECT user.id AS user_id, user.a AS user_a
FROM user1.user
但是实际上，按照
https://www.oschina.net/question/237257_37964 提供的方案，模型更改为
class User(db.Model):
__tablename__ = 'user1.user'
__table_args__ = {'quote': False}
id = Column(BigInteger, primary_key=True)
a = Column(String)
实际上程序给数据库发送的 SQL 代码是
SELECT user1.user.id AS user1.user_id, user1.user.a AS user1.user_a
FROM user1.user
这样的代码是获取不到数据的
还需要什么东西才能让程序发送正确的代码？
SELECT user.id AS user_id, user.a AS user_a
FROM user1.user
Python中如何在Flask项目中使用SQLAlchemy实现Oracle多用户访问？
4 回复
在Flask项目里用SQLAlchemy搞Oracle多用户访问，关键得动态创建引擎和会话。直接用scoped_session配合线程局部变量来管理不同用户的数据库连接。下面是个完整示例：
from flask import Flask, g, request
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
import threading
app = Flask(__name__)
# 用户连接信息存储（实际项目用更安全的方式）
user_connections = {}
def get_user_engine(username, password):
"""为每个用户创建独立的引擎"""
if (username, password) not in user_connections:
# Oracle连接字符串格式
dsn = f'oracle+cx_oracle://{username}:{password}@hostname:1521/service_name'
engine = create_engine(dsn, echo=True)
user_connections[(username, password)] = engine
return user_connections[(username, password)]
# 创建线程局部的session工厂
Session = scoped_session(sessionmaker(), scopefunc=lambda: threading.current_thread().ident)
@app.before_request
def setup_user_session():
"""每个请求前设置用户特定的session"""
# 从请求中获取用户凭证（实际项目应从认证中间件获取）
username = request.headers.get('X-Username')
password = request.headers.get('X-Password')
if username and password:
user_engine = get_user_engine(username, password)
# 绑定session到用户特定的engine
Session.configure(bind=user_engine)
g.db_session = Session
else:
# 默认连接或错误处理
pass
@app.teardown_appcontext
def shutdown_session(exception=None):
"""请求结束后清理session"""
if hasattr(g, 'db_session'):
g.db_session.remove()
@app.route('/query')
def query_data():
"""使用当前用户的session查询数据"""
session = g.db_session
# 执行查询操作
# result = session.query(User).all()
return "Query executed with user-specific connection"
if __name__ == '__main__':
app.run(debug=True)
核心思路就是根据不同的用户凭证创建独立的Engine实例，然后用scoped_session确保每个线程（请求）使用正确的session。注意实际项目中用户凭证应该从安全的认证系统获取，而不是像示例中从请求头直接读取。
用scoped_session管理多用户会话。
