如何在没有任何模型的情况下在blueprint内部的flask sqlalchemy中使用db.execute

Question

我试图在我的蓝图中使用flask_sqlalchemy，而不编写任何模型。我不确定这是否得到支持，但我的目标是用sqlalchemy替换我编写的现有数据库驱动程序。以下是我的应用程序现在的结构

my db.py file

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

my main.py file

from .db import db
if os.getenv('FLASK_ENV') != 'production':
load_dotenv(verbose=True)
app = Flask(__name__)
database_uri = 'postgresql+psycopg2://{user}:{password}@{host}/{database}??sslmode={sslmode}'.format(
user=os.getenv('PGUSER'),
password=os.getenv("PGPASSWORD"), 
host=os.getenv("PGHOST"), 
port=os.getenv("PGPORT"), 
database=os.getenv("PGDATABASE"), 
sslmode=os.getenv("PG_SSL")
)
app.config.update(
SQLALCHEMY_DATABASE_URI=database_uri,
SQLALCHEMY_TRACK_MODIFICATIONS=False,
)
db.init_app(app)
app.register_blueprint(api, urlprefix="/")
cors = CORS(app, supports_credentials=True)
app.secret_key = '079bc80c-cbd6-4fc5-b1aa-8fe01fde50dc'

然后我尝试以以下方式在我的模块中使用

from flask import current_app as app
rows = db.engine.execute("select users.username, users.id, string_agg(coalesce(keys.service, ''), ',') from users left join keys on keys.user_id = users.id where users.username = '" + username + "' and users.password='" + password + "'" + "group by (users.username,users.id)")

但在执行代码时，我的代码中出现了以下错误

RuntimeError: Working outside of application context.
This typically means that you attempted to use functionality that needed
to interface with the current application object in some way. To solve
this, set up an application context with app.app_context().  

我还尝试将我的main.py代码更改为以下

app = Flask(__name__)
database_uri = 'postgresql+psycopg2://{user}:{password}@{host}/{database}??sslmode={sslmode}'.format(
user=os.getenv('PGUSER'),
password=os.getenv("PGPASSWORD"), 
host=os.getenv("PGHOST"), 
port=os.getenv("PGPORT"), 
database=os.getenv("PGDATABASE"), 
sslmode=os.getenv("PG_SSL")
)
app.config.update(
SQLALCHEMY_DATABASE_URI=database_uri,
SQLALCHEMY_TRACK_MODIFICATIONS=False,
)
with app.app_context():
db.init_app(app)
from flask.blueprints import Blueprint
from app.api.routes import api 
# from werkzeug.middleware.profiler import ProfilerMiddleware
# app.wsgi_app = ProfilerMiddleware(app.wsgi_app, restrictions=[5], profile_dir='./profile')
app.register_blueprint(api, urlprefix="/")
cors = CORS(app, supports_credentials=True)
app.secret_key = '079bc80c-cbd6-4fc5-b1aa-8fe01fde50dc'

但我在做时收到了以下错误

AttributeError: 'Flask' object has no attribute 'db'

以我想要的方式执行代码的正确方法是什么？

Answer 1

我刚刚使用了您的代码，并对其进行了一些修改，以在这里创建一个有效的解决方案：https://github.com/amucunguzi/SO-raw-sql-flask-sqlalchemy.

这个CCD_ 3是典型的，当你试图做一些需要CCD_；当前(HTTP(请求"；即整个程序执行不是从view功能启动的(https://flask.palletsprojects.com/en/1.1.x/api/#flask.current_app)。

话虽如此，请确保正确导入模块(请参阅views.py文件(。此外，最好将db作为构造函数依赖项传递给您的模块/服务。