From c3fd66faea8e343ce40e5a749dc0d4250a3974b4 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 use a database session context Drop TODO from the example README --- examples/sqlalchemy/README.rst | 2 - examples/sqlalchemy/app.py | 72 +++++++++++++++++----------------- examples/sqlalchemy/orm.py | 21 ++++++---- 3 files changed, 51 insertions(+), 44 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..7d04bc68c --- a/examples/sqlalchemy/app.py +++ b/examples/sqlalchemy/app.py @@ -5,59 +5,61 @@ 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_session_factory() 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_session_factory() 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_session_factory() 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.now(datetime.UTC) + 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_session_factory() 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.delete(pet) + db_session.commit() + return NoContent, 204 + else: + return NoContent, 404 logging.basicConfig(level=logging.INFO) -db_session = orm.init_db("sqlite:///:memory:") +db_session_factory = orm.init_db() +pets = { + 1: {"name": "Aldo", "animal_type": "cat"}, + 2: {"name": "Bailey", "animal_type": "dog"}, + 3: {"name": "Hugo", "animal_type": "cat"}, +} +for id_, pet in pets.items(): + put_pet(id_, pet) 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..5de062ebf 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,17 @@ 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) +def init_db(): + """ + Initialize the database and return a sessionmaker object. + `check_same_thread` and `StaticPool` are helpful for unit testing of + in-memory sqlite databases; they should not be used in production. + 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.query = db_session.query_property() Base.metadata.create_all(bind=engine) - return db_session + return sessionmaker(autocommit=False, autoflush=False, bind=engine)