Flask中SQLAlchemy automap_base无法映射类的问题求助
问题描述
开发Flask+SQLite的API时,能正常查询数据库架构,但通过Base.classes.Wars声明类对象时触发异常,访问路由时出现NameError: name 'warsTable' is not defined,此前还遇到过AttributeError: 'function' object has no attribute 'WarID'错误。
app.py代码
import numpy as np import sqlalchemy from sqlalchemy.ext.automap import automap_base from sqlalchemy.orm import Session from sqlalchemy import create_engine, func from flask import Flask, jsonify ################################################# # Database Setup ################################################# try: engine = create_engine("sqlite:///static/data/iWars.sqlite") # reflect an existing database into a new model Base = automap_base() # reflect the tables Base.prepare(autoload_with=engine) print("All about that base") print(Base) # # Save reference to the table # tribes = Base.classes.Tribes warsTable = Base.classes.Wars print("yes no maybe") except Exception as e: print("Hey what's up looks like something gnarly happened") print(e) ################################################# # Flask Setup ################################################# app = Flask(__name__) ################################################# # Flask Routes ################################################# @app.route("/") def welcome(): """List all available api routes.""" return ( f"Available Routes:<br/>" # f"/api/v1.0/tribes<br/>" f"/api/v1.0/listwars<br/>" ) @app.route("/api/v1.0/listwars") def warsRoute(): # Create our session (link) from Python to the DB session = Session(engine) # Query all results = session.query(warsTable.WarID).all() session.close() # Create a dictionary from the row data and append to a list of all_passengers all_passengers = [] for id in results: wars_dict = {} wars_dict["id"] = id all_passengers.append(wars_dict) return jsonify(all_passengers) if __name__ == '__main__': app.run(debug=True)
运行日志
执行到warsTable = Base.classes.Wars时进入异常块,仅输出Wars,访问路由时的错误日志:
(mlenv) ianmacsmacbook:USIndigenousWars ianmacmoore$ python app.py All about that base <class 'sqlalchemy.ext.automap.Base'> Hey what's up looks like something gnarly happened Wars Serving Flask app "app" (lazy loading) * Environment: production WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead. * Debug mode: on * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit) * Restarting with watchdog (fsevents) All about that base <class 'sqlalchemy.ext.automap.Base'> Hey what's up looks like something gnarly happened Wars * Debugger is active! * Debugger PIN: 955-667-953 127.0.0.1 - - [11/Mar/2023 13:19:44] "GET / HTTP/1.1" 200 - 127.0.0.1 - - [11/Mar/2023 13:19:54] "GET /api/v1.0/listwars HTTP/1.1" 500 - Traceback (most recent call last): File "/opt/anaconda3/envs/mlenv/lib/python3.7/site-packages/flask/app.py", line 2464, in __call__ return self.wsgi_app(environ, start_response) File "/opt/anaconda3/envs/mlenv/lib/python3.7/site-packages/flask/app.py", line 2450, in wsgi_app response = self.handle_exception(e) File "/opt/anaconda3/envs/mlenv/lib/python3.7/site-packages/flask/app.py", line 1867, in handle_exception reraise(exc_type, exc_value, tb) File "/opt/anaconda3/envs/mlenv/lib/python3.7/site-packages/flask/_compat.py", line 39, in reraise raise value File "/opt/anaconda3/envs/mlenv/lib/python3.7/site-packages/flask/app.py", line 2447, in wsgi_app response = self.full_dispatch_request() File "/opt/anaconda3/envs/mlenv/lib/python3.7/site-packages/flask/app.py", line 1952, in full_dispatch_request rv = self.handle_user_exception(e) File "/opt/anaconda3/envs/mlenv/lib/python3.7/site-packages/flask/app.py", line 1821, in handle_user_exception reraise(exc_type, exc_value, tb) File "/opt/anaconda3/envs/mlenv/lib/python3.7/site-packages/flask/_compat.py", line 39, in reraise raise value File "/opt/anaconda3/envs/mlenv/lib/python3.7/site-packages/flask/app.py", line 1950, in full_dispatch_request rv = self.dispatch_request() File "/opt/anaconda3/envs/mlenv/lib/python3.7/site-packages/flask/app.py", line 1936, in dispatch_request return self.view_functions[rule.endpoint](**req.view_args) File "/Users/ianmacmoore/Documents/BluePlusRed/USIndigenousWars/app.py", line 56, in warsRoute results = session.query(warsTable.WarID).all() NameError: name 'warsTable' is not defined
最初写入数据库的类定义
class Against(Base): __tablename__ = 'Against' TribeID = Column(Integer, primary_key=True) AgainstCode = Column(Integer) WarID = Column(Integer) class Tribes(Base): __tablename__ = 'Tribes' TribeID = Column(Integer, primary_key=True) TribeName = Column(String(255)) TribeName2 = Column(String(255)) TribeName3 = Column(String(255)) class Wars(Base): __tablename__ = 'Wars' WarID = Column(Integer, primary_key=True) WarName = Column(String(255)) StartYear = Column(Integer) EndYear = Column(Integer) WikiLink = Column(String(255)) LengthYears = Column(Integer) class YearSum(Base): __tablename__ = 'YearSum' Year = Column(Integer, primary_key=True) SumWars = Column(Integer) y = Column(Integer)
Jupyter Notebook中可正常创建表对象
wars = Table("Wars", metadata_obj, autoload_with=engine) wars
输出:
Table('Wars', MetaData(bind=None), Column('index', BIGINT(), table=<Wars>), Column('WarID', BIGINT(), table=<Wars>), Column('War Name', TEXT(), table=<Wars>), Column('StartYear', BIGINT(), table=<Wars>), Column('EndYear', BIGINT(), table=<Wars>), Column('WikiLink', TEXT(), table=<Wars>), Column('LengthYears', BIGINT(), table=<Wars>), schema=None)
当前运行环境的相关包版本
conda list '^(python|flask|sqlal)' # packages in environment at /opt/anaconda3/envs/mlenv: # # Name Version Build Channel flask 1.1.2 pyhd3eb1b0_0 python 3.7.13 hdfd78df_0 python-dateutil 2.8.2 pyhd3eb1b0_0 python-fastjsonschema 2.16.2 py37hecd8cb5_0 python-libarchive-c 2.9 pyhd3eb1b0_1 python-lsp-black 1.2.1 py37hecd8cb5_0 python-lsp-jsonrpc 1.0.0 pyhd3eb1b0_0 python-lsp-server 1.5.0 py37hecd8cb5_0 python-slugify 5.0.2 pyhd3eb1b0_0 python-snappy 0.6.0 py37h23ab428_3 python.app 3 py37hca72f7f_0 sqlalchemy 1.4.39 py37hca72f7f_0
解决方案
修复automap表名映射问题
SQLAlchemy的automap_base反射表时,默认会将表名转为小写(比如Wars会映射为wars),需用Base.classes.wars而非Base.classes.Wars获取映射类。优化异常处理逻辑
当前try-except块会跳过变量定义导致后续路由报错,建议先打印所有反射成功的表类,确认名称:print("反射得到的表类:", list(Base.classes.keys()))同时在初始化失败时抛出异常,避免程序带病运行:
except Exception as e: print("数据库初始化失败:", str(e)) raise修改后的数据库初始化代码
try:
engine = create_engine("sqlite:///static/data/iWars.sqlite")
Base = automap_base() Base.prepare(autoload_with=engine) print("反射得到的表类:", list(Base.classes.keys())) warsTable = Base.classes.wars print("成功获取Wars表类")
except Exception as e:
print("数据库初始化失败:", str(e))
raise
4. **修复路由查询结果处理** 查询返回的`results`是元组列表,需提取元组内的具体值: ```python all_passengers = [] for id_tuple in results: wars_dict = {} wars_dict["id"] = id_tuple[0] all_passengers.append(wars_dict)
补充说明
Jupyter中用Table直接加载表能正常工作,是因为Table基于元数据直接读取,而automap_base会自动转换表名格式,这是两者的核心差异。
内容的提问来源于stack exchange,提问作者Mac
相关产品推荐
相关产品推荐

