python - How to map SQLAlchemy raw text query result to model instances? -
here's example flask app illustrates i'm trying do:
"""flask app list yummy desserts.""" # these dependencies: # flask 0.12.2 # sqlalchemy 1.1.13 # flask-sqlalchemy 2.2 # sqlalchemy-utils 0.32.14 # alembic 0.9.5 # flask-migrate 2.1.0 import click flask import flask flask.cli import with_appcontext flask_migrate import migrate flask_sqlalchemy import sqlalchemy sqlalchemy import orm, select, text sqlalchemy_utils import choicetype # use app, place file in './myproject/app.py', , # create file './autoapp.py' looks this: # myproject.app import create_app, config # app = create_app(config) # , set: # export flask_app=/path/to/autoapp.py class config(object): """configuration object.""" debug = true secret_key = 'secret-key' sqlalchemy_database_uri = 'postgresql://quicktest:quicktest@localhost/quicktest' sqlalchemy_track_modifications = false db = sqlalchemy() migrate = migrate() column = db.column topping_types = [ ('chocolate', 'chocolate'), ('caramel', 'caramel'), ('honeycomb', 'honeycomb'), ] # example db models # init db with: # flask db init # flask db migrate # edit migration file, , add imports section: # import sqlalchemy_utils # , edit 'topping_type' be: # sa.column( # 'topping_type', # sqlalchemy_utils.types.choice.choicetype([('', '')]), # nullable=false), class dessert(db.model): __tablename__ = 'dessert' id = column(db.integer, primary_key=true) topping_type = column(choicetype(topping_types), nullable=false) dessert_revisions = db.relationship( 'dessertrevision', backref=db.backref('dessert'), cascade='all, delete-orphan', lazy='dynamic') class dessertrevision(db.model): __tablename__ = 'dessert_revision' id = column(db.integer, primary_key=true) name = column(db.string(255), nullable=false, default='') dessert_id = column(db.integer, db.foreignkey(dessert.id), nullable=false) @click.command() @with_appcontext def create_desserts(): """create desserts.""" d1 = dessert(topping_type='chocolate') dr1 = dessertrevision(dessert=d1, name='mouth-watering tiramisu') dr2 = dessertrevision(dessert=d1, name='gut-wrenching tiramisu') db.session.add(d1) db.session.add(dr1) db.session.add(dr2) d2 = dessert(topping_type='honeycomb') dr3 = dessertrevision(dessert=d2, name='honey-soaked waffles') db.session.add(d2) db.session.add(dr3) db.session.commit() click.echo('created yummy desserts') @click.command() @with_appcontext def list_chocolate_desserts_select_column_attrs(): """list chocolate desserts selecting columns attributes.""" query = ( select([ dessert.id, dessert.topping_type, dessertrevision.id, dessertrevision.name, dessertrevision.dessert_id], use_labels=true) .select_from(dessertrevision.__table__.join(dessert)) .where(dessert.topping_type=='chocolate')) result = (db.session .query(dessert, dessertrevision) .from_statement(query)) # instead of above, can this, achieves same: # conn = db.session.connection() # cursor = conn.execute(query) # result = (db.session # .query(dessert, dessertrevision) # .instances(cursor)) dessert, dr in result: print({ 'dessert_id': dessert.id, # works nicely, outputs: # 'topping_type': 'chocolate' # i.e. processes raw cursor row, before populating # db model object row's values 'topping_type': ( dessert.topping_type.code if dessert.topping_type else none), 'dessert_revision_id': dr.id, 'name': dr.name}) @click.command() @with_appcontext def list_chocolate_desserts_select_column_strings(): """list chocolate desserts selecting columns strings.""" query = ( select([ text('dessert.id dessert_id'), text('dessert.topping_type dessert_topping_type'), text('dessert_revision.id dessert_revision_id'), text('dessert_revision.name dessert_revision_name'), text('dessert_revision.dessert_id dessert_revision_dessert_id')]) .select_from(dessertrevision.__table__.join(dessert)) .where(dessert.topping_type=='chocolate')) # tried (in various combinations): # mapper = orm.class_mapper(dessert) # conn = db.session.connection( # mapper=mapper, clause=query, close_with_result=true) # cursor = conn.execute(query) # result = (db.session # .query(dessert, dessertrevision) # .instances(cursor)) # doesn't help, still doesn't process row values result = (db.session .query(dessert, dessertrevision) .from_statement(query)) dessert, dr in result: print({ 'dessert_id': dessert.id, # doesn't work intended, raises: # attributeerror: 'str' object has no attribute 'code' # i.e. doesn't process raw cursor row, db model # object gets populated row's values # they're stored in db , returned db driver 'topping_type': ( dessert.topping_type.code if dessert.topping_type else none), 'dessert_revision_id': dr.id, 'name': dr.name}) @click.command() @with_appcontext def list_chocolate_desserts_raw_sql(): """list chocolate desserts raw sql.""" query = ( 'select dessert.id dessert_id, ' 'dessert.topping_type dessert_topping_type, ' 'dessert_revision.id dessert_revision_id, ' 'dessert_revision.name dessert_revision_name, ' 'dessert_revision.dessert_id dessert_revision_dessert_id ' 'from dessert_revision ' 'join dessert ' 'on dessert_revision.dessert_id = dessert.id ' "where dessert.topping_type = 'chocolate'") # tried (in various combinations): # mapper = orm.class_mapper(dessert) # conn = db.session.connection( # mapper=mapper, clause=query, close_with_result=true) # cursor = conn.execute(query) # result = (db.session # .query(dessert, dessertrevision) # .instances(cursor)) # doesn't help, still doesn't process row values result = (db.session .query(dessert, dessertrevision) .from_statement(query)) dessert, dr in result: print({ 'dessert_id': dessert.id, # doesn't work intended, raises: # attributeerror: 'str' object has no attribute 'code' # i.e. doesn't process raw cursor row, db model # object gets populated row's values # they're stored in db , returned db driver 'topping_type': ( dessert.topping_type.code if dessert.topping_type else none), 'dessert_revision_id': dr.id, 'name': dr.name}) def create_app(config_object=config): """an application factory.""" app = flask(__name__) app.config.from_object(config_object) db.init_app(app) migrate.init_app(app, db) app.cli.add_command(create_desserts) app.cli.add_command(list_chocolate_desserts_select_column_attrs) app.cli.add_command(list_chocolate_desserts_select_column_strings) app.cli.add_command(list_chocolate_desserts_raw_sql) return app
in actual app, have large raw text query i'm executing directly, , i'm doing declarative sqlalchemy query load model objects, based on id values returned text query (i.e. calling separate queries after main query, of form dessert.query.filter(dessert.id.in_(dessert_ids)).all()
, each type of model i'm querying). have model objects results queried using raw text, , have access conveniences associated model objects.
i avoid multiple sql queries need in order achieve - ideally, assuming list fields of models in raw text query, i'd list of declarative model objects directly, 1 query.
so, in example app above, i'm trying (but fails) in list_chocolate_desserts_raw_sql()
function. stepping stone if use (but fails) list_chocolate_desserts_select_column_strings()
function. however, 1 works list_chocolate_desserts_select_column_attrs()
function (and in function, want - i.e. gives me model instances, , per debug logs, executes 1 sql query in order achieve it).
this similar question asked @ sqlalchemy model instance query / cursor results - and, testing, answer suggested there doesn't work (it gave raw text query example).
the problem happens if, of fields given model, they're of type raw value needs processed value model gets populated with. e.g. sqlalchemy-utils
choicetype
field in example (same problem various other field types, e.g. uuids stored binary values in db). in guts of sqlalchemy (in resultproxy
, rowproxy
, far can tell), processing supposed happen while fetching values db cursor object, , before passing values on declarative layer.
so, question: there way achieve i'm trying raw text query? or viable option convert query proper sqlalchemy query instance (using actual column attributes), sqlalchemy can find metadata model fields, , can row processing it's supposed do?
Comments
Post a Comment