fix: add pg8000 graceful shutdown on Cloud Run scale-down

legal-api/src/legal_api/__init__.py

@@ -35,12 +35,14 @@
 
 This module is the API for the Legal Entity system.
 """
+import logging
 import os
 from typing import Optional
 
 from flask import Flask, jsonify
 from registry_schemas import __version__ as registry_schemas_version
 from registry_schemas.flask import SchemaServices
+from sqlalchemy import event
 
 from legal_api import config, models
 from legal_api.models import db
@@ -61,6 +63,29 @@
 setup_logging(os.path.join(os.path.abspath(os.path.dirname(__file__)), "logging.conf"))  # important to do this first
 
 
+def _setup_pg8000_graceful_shutdown(engine) -> None:
+    """Suppress pg8000 InterfaceError on connection close during Cloud Run scale-down."""
+    try:
+        from pg8000.exceptions import InterfaceError as _interface_error  # noqa: N813
+    except ImportError:
+        _interface_error = None
+
+    @event.listens_for(engine, "connect")
+    def on_connect(dbapi_conn, _connection_record):
+        orig_close = dbapi_conn.close
+
+        def safe_close():
+            try:
+                orig_close()
+            except Exception as exc:
+                if _interface_error and isinstance(exc, _interface_error):
+                    logging.getLogger(__name__).debug("Suppressed pg8000 InterfaceError on teardown.")
+                else:
+                    raise
+
+        dbapi_conn.close = safe_close
+
+
 def create_app(run_mode: Optional[str] = None, **kwargs) -> Flask:
     """Return a configured Flask App using the Factory method."""
     run_mode = run_mode or os.getenv("RUN_MODE", "production")
@@ -72,18 +97,6 @@ def create_app(run_mode: Optional[str] = None, **kwargs) -> Flask:
     else:
         app.config.from_object(config.CONFIGURATION[run_mode])
 
-    if app.config.get("CLOUDSQL_INSTANCE_CONNECTION_NAME"):  # pragma: no cover
-        from cloud_sql_connector import DBConfig
-        db_config = DBConfig(
-            instance_name=app.config["CLOUDSQL_INSTANCE_CONNECTION_NAME"],
-            database=app.config.get("DB_NAME", ""),
-            user=app.config.get("DB_USER", ""),
-            ip_type=app.config["DB_IP_TYPE"],
-            pool_recycle = 60,
-            schema="public",
-        )
-        app.config["SQLALCHEMY_ENGINE_OPTIONS"] = db_config.get_engine_options()
-
     app.logger = StructuredLogging(app).get_logger()
     init_db(app)
     rsbc_schemas.init_app(app)
@@ -94,6 +107,8 @@ def create_app(run_mode: Optional[str] = None, **kwargs) -> Flask:
 
     with app.app_context():  # db require app context
         digital_credentials.init_app(app)
+        if app.config.get("CLOUDSQL_INSTANCE_CONNECTION_NAME"):  # pragma: no cover
+            _setup_pg8000_graceful_shutdown(db.engine)
 
     cache.init_app(app)
 

legal-api/src/legal_api/config.py

@@ -23,6 +23,7 @@
 import os
 import sys
 
+from cloud_sql_connector import DBConfig
 from dotenv import find_dotenv, load_dotenv
 
 # this will load all the envars from a .env file located in the project root (api)
@@ -118,6 +119,15 @@ class _Config:  # pylint: disable=too-few-public-methods
         SQLALCHEMY_DATABASE_URI = f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
     elif CLOUDSQL_INSTANCE_CONNECTION_NAME:
         SQLALCHEMY_DATABASE_URI = "postgresql+pg8000://"
+        db_config = DBConfig(
+            instance_name=CLOUDSQL_INSTANCE_CONNECTION_NAME,
+            database=DB_NAME,
+            user=DB_USER,
+            ip_type=DB_IP_TYPE,
+            pool_recycle=60,
+            schema="public",
+        )
+        SQLALCHEMY_ENGINE_OPTIONS = db_config.get_engine_options()
 
     # JWT_OIDC Settings
     JWT_OIDC_WELL_KNOWN_CONFIG = os.getenv("JWT_OIDC_WELL_KNOWN_CONFIG")

queue_services/business-bn/src/business_bn/__init__.py

@@ -36,6 +36,7 @@
 """
 import os
 
+from cloud_sql_connector import setup_pg8000_close_event_listener
 from flask import Flask
 
 from business_model.models import db
@@ -57,19 +58,10 @@ def create_app(environment: str = os.getenv("DEPLOYMENT_ENV", "production"), **k
     app.logger = StructuredLogging(app).get_logger()
     app.config.from_object(CONFIGURATION[environment])
 
-    if app.config.get("CLOUDSQL_INSTANCE_CONNECTION_NAME"):  # pragma: no cover
-        from cloud_sql_connector import DBConfig
-        db_config = DBConfig(
-            instance_name=app.config["CLOUDSQL_INSTANCE_CONNECTION_NAME"],
-            database=app.config.get("DB_NAME", ""),
-            user=app.config.get("DB_USER", ""),
-            ip_type=app.config["DB_IP_TYPE"],
-            pool_recycle=60,
-            schema="public",
-        )
-        app.config["SQLALCHEMY_ENGINE_OPTIONS"] = db_config.get_engine_options()
-
     db.init_app(app)
+
+    with app.app_context():
+        setup_pg8000_close_event_listener(db.engine)
     register_endpoints(app)
     gcp_queue.init_app(app)
 

queue_services/business-bn/src/business_bn/config.py

@@ -42,6 +42,7 @@
 
 import os
 
+from cloud_sql_connector import DBConfig
 from dotenv import find_dotenv, load_dotenv
 
 # this will load all the envars from a .env file located in the project root (api)
@@ -93,6 +94,15 @@ class Config:  # pylint: disable=too-few-public-methods
         SQLALCHEMY_DATABASE_URI = f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
     elif CLOUDSQL_INSTANCE_CONNECTION_NAME:
         SQLALCHEMY_DATABASE_URI = "postgresql+pg8000://"
+        db_config = DBConfig(
+            instance_name=CLOUDSQL_INSTANCE_CONNECTION_NAME,
+            database=DB_NAME,
+            user=DB_USER,
+            ip_type=DB_IP_TYPE,
+            pool_recycle=60,
+            schema="public",
+        )
+        SQLALCHEMY_ENGINE_OPTIONS = db_config.get_engine_options()
 
     # legislative timezone for future effective dating
     LEGISLATIVE_TIMEZONE = os.getenv("LEGISLATIVE_TIMEZONE", "America/Vancouver")

queue_services/business-digital-credentials/pyproject.toml

@@ -133,7 +133,49 @@ docstring-quotes = "double"
 "**/__init__.py" = ["F401"] # used for imports
 
 [tool.pytest.ini_options]
-addopts = "--cov=src"
+minversion = "2.0"
+testpaths = [
+   "tests",
+]
+addopts = "--verbose --strict -p no:warnings --cov=src --cov-report html:htmlcov --cov-report xml:coverage.xml"
+python_files = [
+   "test*.py"
+]
+norecursedirs = [
+   ".git", ".tox", "venv*", "requirements*", "build",
+]
+log_cli = true
+log_cli_level = "1"
+filterwarnings = [
+   "ignore::UserWarning"
+]
+markers = [
+   "slow",
+   "serial",
+]
+
+[tool.coverage.run]
+branch = true
+source = [
+   "src/business_digital_credentials",
+]
+omit = [
+    "wsgi.py",
+]
+
+[tool.coverage.report]
+exclude_lines = [
+   "pragma: no cover",
+   "from",
+   "import",
+   "def __repr__",
+   "if self.debug:",
+   "if settings.DEBUG",
+   "raise AssertionError",
+   "raise NotImplementedError",
+   "if 0:",
+   'if __name__ == "__main__":',
+]
 
 [build-system]
 requires = ["poetry-core>=2.0.0,<3.0.0"]

queue_services/business-digital-credentials/src/business_digital_credentials/__init__.py

@@ -18,7 +18,7 @@
 import os
 
 from business_registry_digital_credentials import digital_credentials
-from cloud_sql_connector import DBConfig
+from cloud_sql_connector import setup_pg8000_close_event_listener
 from flask import Flask
 
 from business_model.models.db import db
@@ -50,22 +50,12 @@ def create_app(
     if app.config.get("LD_SDK_KEY", None):
         flags.init_app(app)
 
-    if app.config.get("CLOUDSQL_INSTANCE_CONNECTION_NAME"):  # pragma: no cover
-        db_config = DBConfig(
-            instance_name=app.config["CLOUDSQL_INSTANCE_CONNECTION_NAME"],
-            database=app.config.get("DB_NAME", ""),
-            user=app.config.get("DB_USER", ""),
-            ip_type=app.config["DB_IP_TYPE"],
-            pool_recycle=60,
-            schema="public",
-        )
-        app.config["SQLALCHEMY_ENGINE_OPTIONS"] = db_config.get_engine_options()
-
     db.init_app(app)
     register_endpoints(app)
     gcp_queue.init_app(app)
 
     with app.app_context():
         digital_credentials.init_app(app)
+        setup_pg8000_close_event_listener(db.engine)
 
     return app

queue_services/business-digital-credentials/src/business_digital_credentials/config.py

@@ -22,6 +22,7 @@
 
 import os
 
+from cloud_sql_connector import DBConfig
 from dotenv import find_dotenv, load_dotenv
 
 # this will load all the envars from a .env file located in the project root (api)
@@ -80,6 +81,15 @@ class Config:  # pylint: disable=too-few-public-methods
         SQLALCHEMY_DATABASE_URI = f"postgresql+pg8000://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
     elif CLOUDSQL_INSTANCE_CONNECTION_NAME:
         SQLALCHEMY_DATABASE_URI = "postgresql+pg8000://"
+        db_config = DBConfig(
+            instance_name=CLOUDSQL_INSTANCE_CONNECTION_NAME,
+            database=DB_NAME,
+            user=DB_USER,
+            ip_type=DB_IP_TYPE,
+            pool_recycle=60,
+            schema="public",
+        )
+        SQLALCHEMY_ENGINE_OPTIONS = db_config.get_engine_options()
 
     # Traction ACA-Py tenant settings to issue credentials from
     TRACTION_API_URL = os.getenv("TRACTION_API_URL")

queue_services/business-emailer/pyproject.toml

@@ -200,7 +200,49 @@ docstring-quotes = "double"
 "src/business_emailer/services/namex.py" = ["I001"]  # ignoring 'unordered' imports
 
 [tool.pytest.ini_options]
-addopts = "--cov=src"
+minversion = "2.0"
+testpaths = [
+   "tests",
+]
+addopts = "--verbose --strict -p no:warnings --cov=src --cov-report html:htmlcov --cov-report xml:coverage.xml"
+python_files = [
+   "test*.py"
+]
+norecursedirs = [
+   ".git", ".tox", "venv*", "requirements*", "build",
+]
+log_cli = true
+log_cli_level = "1"
+filterwarnings = [
+   "ignore::UserWarning"
+]
+markers = [
+   "slow",
+   "serial",
+]
+
+[tool.coverage.run]
+branch = true
+source = [
+   "src/business_emailer",
+]
+omit = [
+    "wsgi.py",
+]
+
+[tool.coverage.report]
+exclude_lines = [
+   "pragma: no cover",
+   "from",
+   "import",
+   "def __repr__",
+   "if self.debug:",
+   "if settings.DEBUG",
+   "raise AssertionError",
+   "raise NotImplementedError",
+   "if 0:",
+   'if __name__ == "__main__":',
+]
 
 [build-system]
 requires = ["poetry-core>=2.0.0,<3.0.0"]

queue_services/business-emailer/src/business_emailer/__init__.py

@@ -15,6 +15,7 @@
 
 This module is the service worker for sending emails about entity related events.
 """
+from cloud_sql_connector import setup_pg8000_close_event_listener
 from flask import Flask
 
 from business_model.models.db import db
@@ -35,19 +36,10 @@ def create_app(environment: Config = ProdConfig, **kwargs) -> Flask:
     if app.config.get("LD_SDK_KEY", None):
         flags.init_app(app)
 
-    if app.config.get("CLOUDSQL_INSTANCE_CONNECTION_NAME"):  # pragma: no cover
-        from cloud_sql_connector import DBConfig
-        db_config = DBConfig(
-            instance_name=app.config["CLOUDSQL_INSTANCE_CONNECTION_NAME"],
-            database=app.config.get("DB_NAME", ""),
-            user=app.config.get("DB_USER", ""),
-            ip_type=app.config["DB_IP_TYPE"],
-            pool_recycle=60,
-            schema="public",
-        )
-        app.config["SQLALCHEMY_ENGINE_OPTIONS"] = db_config.get_engine_options()
-
     db.init_app(app)
+
+    with app.app_context():
+        setup_pg8000_close_event_listener(db.engine)
     register_endpoints(app)
     gcp_queue.init_app(app)