From 8430e1aa8a572f202713257850cdd6c3ec4887a9 Mon Sep 17 00:00:00 2001 From: Christopher Lott Date: Thu, 5 Dec 2024 16:48:34 -0500 Subject: [PATCH] Update example for latest version of SQLAlchemy Pass argument that disables SQLite thread check when creating the engine Revise request-processing functions to work within a database session context Drop TODO from the example README --- examples/sqlalchemy/README.rst | 2 -- examples/sqlalchemy/app.py | 65 ++++++++++++++++------------------ examples/sqlalchemy/orm.py | 17 ++++----- 3 files changed, 39 insertions(+), 45 deletions(-) mode change 100755 => 100644 examples/sqlalchemy/app.py diff --git a/examples/sqlalchemy/README.rst b/examples/sqlalchemy/README.rst index 35b20097c..cf40d1c1e 100644 --- a/examples/sqlalchemy/README.rst +++ b/examples/sqlalchemy/README.rst @@ -4,8 +4,6 @@ SQLAlchemy Example .. note:: - TODO: Update this example to work with recent (2024) versions of Python and SQLAlchemy. - A simple example of how one might use SQLAlchemy as a backing store for a Connexion based application. diff --git a/examples/sqlalchemy/app.py b/examples/sqlalchemy/app.py old mode 100755 new mode 100644 index 04870d96d..9c48ed092 --- a/examples/sqlalchemy/app.py +++ b/examples/sqlalchemy/app.py @@ -5,59 +5,54 @@ import orm from connexion import NoContent -db_session = None - def get_pets(limit, animal_type=None): - q = db_session.query(orm.Pet) - if animal_type: - q = q.filter(orm.Pet.animal_type == animal_type) - return [p.dump() for p in q][:limit] + with db_sess_mkr() as db_session: + q = db_session.query(orm.Pet) + if animal_type: + q = q.filter(orm.Pet.animal_type == animal_type) + return [p.dump() for p in q][:limit] def get_pet(pet_id): - pet = db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).one_or_none() - return pet.dump() if pet is not None else ("Not found", 404) + with db_sess_mkr() as db_session: + pet = db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).one_or_none() + return pet.dump() if pet is not None else ("Not found", 404) def put_pet(pet_id, pet): - p = db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).one_or_none() - pet["id"] = pet_id - if p is not None: - logging.info("Updating pet %s..", pet_id) - p.update(**pet) - else: - logging.info("Creating pet %s..", pet_id) - pet["created"] = datetime.datetime.utcnow() - db_session.add(orm.Pet(**pet)) - db_session.commit() - return NoContent, (200 if p is not None else 201) + with db_sess_mkr() as db_session: + p = db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).one_or_none() + pet["id"] = pet_id + if p is not None: + logging.info("Updating pet %s..", pet_id) + p.update(**pet) + else: + logging.info("Creating pet %s..", pet_id) + pet["created"] = datetime.datetime.utcnow() + db_session.add(orm.Pet(**pet)) + db_session.commit() + return NoContent, (200 if p is not None else 201) def delete_pet(pet_id): - pet = db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).one_or_none() - if pet is not None: - logging.info("Deleting pet %s..", pet_id) - db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).delete() - db_session.commit() - return NoContent, 204 - else: - return NoContent, 404 + with db_sess_mkr() as db_session: + pet = db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).one_or_none() + if pet is not None: + logging.info("Deleting pet %s..", pet_id) + db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).delete() + db_session.commit() + return NoContent, 204 + else: + return NoContent, 404 logging.basicConfig(level=logging.INFO) -db_session = orm.init_db("sqlite:///:memory:") +db_sess_mkr = orm.init_db() app = connexion.FlaskApp(__name__, specification_dir="spec") app.add_api("openapi.yaml") app.add_api("swagger.yaml") -application = app.app - - -@application.teardown_appcontext -def shutdown_session(exception=None): - db_session.remove() - if __name__ == "__main__": app.run(port=8080, reload=False) diff --git a/examples/sqlalchemy/orm.py b/examples/sqlalchemy/orm.py index 52c4c6f79..7560da673 100644 --- a/examples/sqlalchemy/orm.py +++ b/examples/sqlalchemy/orm.py @@ -1,6 +1,7 @@ from sqlalchemy import Column, DateTime, String, create_engine from sqlalchemy.ext.declarative import declarative_base -from sqlalchemy.orm import scoped_session, sessionmaker +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool Base = declarative_base() @@ -24,11 +25,11 @@ def dump(self): return {k: v for k, v in vars(self).items() if not k.startswith("_")} -def init_db(uri): - engine = create_engine(uri, convert_unicode=True) - db_session = scoped_session( - sessionmaker(autocommit=False, autoflush=False, bind=engine) - ) - Base.query = db_session.query_property() +def init_db(): + # https://stackoverflow.com/questions/6519546/scoped-sessionsessionmaker-or-plain-sessionmaker-in-sqlalchemy + engine = create_engine( + url="sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool) Base.metadata.create_all(bind=engine) - return db_session + return sessionmaker(autocommit=False, autoflush=False, bind=engine)