Update example for latest version of SQLAlchemy

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.
 

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)

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)