20346 - Refactor Oracle to point at Datawarehouse instead of CPRD (#1804

)

jobs/payment-jobs/config.py

@@ -73,11 +73,13 @@ class _Config(object):  # pylint: disable=too-few-public-methods
         SQLALCHEMY_DATABASE_URI = f"postgresql+pg8000://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{int(DB_PORT)}/{DB_NAME}"
     SQLALCHEMY_ECHO = False
 
-    ORACLE_USER = os.getenv("ORACLE_USER", "")
-    ORACLE_PASSWORD = os.getenv("ORACLE_PASSWORD", "")
-    ORACLE_DB_NAME = os.getenv("ORACLE_DB_NAME", "")
-    ORACLE_HOST = os.getenv("ORACLE_HOST", "")
-    ORACLE_PORT = int(os.getenv("ORACLE_PORT", "1521"))
+    # Data Warehouse Settings
+    DW_UNIX_SOCKET = os.getenv("DW_UNIX_SOCKET", "")
+    DW_NAME = os.getenv("DW_NAME", "")
+    DW_USER = os.getenv("DW_USER", "")
+    DW_PASSWORD = os.getenv("DW_PASSWORD", "")
+    DW_HOST = os.getenv("DW_HOST", "")
+    DW_PORT = os.getenv("DW_PORT", "5432")
 
     # PAYBC Direct Pay Settings
     PAYBC_DIRECT_PAY_REF_NUMBER = os.getenv("PAYBC_DIRECT_PAY_REF_NUMBER")

jobs/payment-jobs/devops/vaults.gcp.env

@@ -81,8 +81,9 @@ AUTH_WEB_STATEMENT_URL="op://web-url/$APP_ENV/auth-web/AUTH_WEB_URL"
 REGISTRIES_LOGO_IMAGE_NAME="bc_logo_for_email.png"
 DISBURSEMENT_DELAY="5"
 DISABLE_CFS_FAS_INTEGRATION="false"
-ORACLE_USER=""
-ORACLE_PASSWORD=""
-ORACLE_DB_NAME=""
-ORACLE_HOST=""
-ORACLE_PORT="1521"
+DW_UNIX_SOCKET="op://database/$APP_ENV/fin-warehouse/DATABASE_UNIX_SOCKET"
+DW_NAME="op://database/$APP_ENV/fin-warehouse/DATABASE_NAME"
+DW_USER="op://database/$APP_ENV/fin-warehouse/DB_READONLY_USER"
+DW_PASSWORD="op://database/$APP_ENV/fin-warehouse/DB_READONLY_PASSWORD"
+DW_HOST="op://database/$APP_ENV/fin-warehouse/DATABASE_HOST"
+DW_PORT="op://database/$APP_ENV/fin-warehouse/DATABASE_PORT"

jobs/payment-jobs/devops/vaults.json

@@ -66,5 +66,11 @@
             "gtksf3",
             "topics"
         ]
+    },
+    {
+        "vault": "database",
+        "application": [
+            "fin-warehouse"
+        ]
     }
 ]

jobs/payment-jobs/invoke_jobs.py

@@ -25,7 +25,7 @@
 from sentry_sdk.integrations.flask import FlaskIntegration
 
 import config
-from services import oracle_db
+from services import data_warehouse
 from tasks.eft_overpayment_notification_task import EFTOverpaymentNotificationTask
 from tasks.eft_statement_due_task import EFTStatementDueTask
 from tasks.eft_task import EFTTask
@@ -38,7 +38,7 @@
 def create_app(
     run_mode=os.getenv("DEPLOYMENT_ENV", "production"),
     job_name="unknown",
-    init_oracle=False,
+    init_data_warehouse=False,
 ):
     """Return a configured Flask App using the Factory method."""
     from pay_api.models import db, ma
@@ -58,8 +58,9 @@ def create_app(
     app.logger.info("<<<< Starting Payment Jobs >>>>")
     queue.init_app(app)
     db.init_app(app)
-    if init_oracle:
-        oracle_db.init_app(app)
+    if init_data_warehouse:
+        data_warehouse.init_app(app)
+
     ma.init_app(app)
     flag_service = Flags()
     flag_service.init_app(app)
@@ -94,8 +95,8 @@ def run(job_name, argument=None):
     from tasks.statement_task import StatementTask
     from tasks.unpaid_invoice_notify_task import UnpaidInvoiceNotifyTask
 
-    jobs_with_oracle_connections = ["BCOL_REFUND_CONFIRMATION"]
-    application = create_app(job_name=job_name, init_oracle=job_name in jobs_with_oracle_connections)
+    jobs_with_data_warehouse_connections = ["BCOL_REFUND_CONFIRMATION"]
+    application = create_app(job_name=job_name, init_data_warehouse=job_name in jobs_with_data_warehouse_connections)
 
     application.app_context().push()
     match job_name:

jobs/payment-jobs/pyproject.toml

@@ -25,9 +25,9 @@ flask-migrate = "^4.0.7"
 itsdangerous = "^2.1.2"
 dataclass-wizard = "^0.22.3"
 launchdarkly-server-sdk = "^8.2.1"
-cx-oracle = "^8.3.0"
 more-itertools = "^10.2.0"
 pg8000 = "^1.30.5"
+cloud-sql-python-connector = "^1.13.0"
 
 
 [tool.poetry.group.dev.dependencies]

jobs/payment-jobs/services/__init__.py

@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """This module holds services used by jobs."""
-from .oracle import oracle_db
+from .data_warehouse import data_warehouse

jobs/payment-jobs/services/data_warehouse.py

@@ -0,0 +1,116 @@
+# Copyright © 2022 Province of British Columbia
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Create data warehouse connection.
+
+These will get initialized by the application.
+"""
+
+# services/data_warehouse.py
+
+import pg8000
+from sqlalchemy import create_engine
+from sqlalchemy.orm import scoped_session, sessionmaker
+from google.cloud.sql.connector import Connector
+from dataclasses import dataclass
+
+
+@dataclass
+class DBConfig:
+    """Database configuration settings."""
+
+    database: str
+    user: str
+    password: str
+    unix_sock: str = None
+    host: str = None  # Optional for TCP connection
+    port: int = 5432  # Optional with default for PostgreSQL
+
+
+def getconn(connector: Connector, db_config: DBConfig) -> object:
+    """Create a database connection.
+
+    Args:
+        connector (Connector): The Google Cloud SQL connector instance.
+        db_config (DBConfig): The database configuration.
+
+    Returns:
+        object: A connection object to the database.
+    """
+    if db_config.unix_sock:
+        # Use Unix socket connection with the Connector for deployment
+        instance_connection_string = db_config.unix_sock.replace('/cloudsql/', '')
+        return connector.connect(
+            instance_connection_string=instance_connection_string,
+            ip_type='private',
+            user=db_config.user,
+            password=db_config.password,
+            db=db_config.database,
+            driver='pg8000',
+        )
+    else:
+        conn = pg8000.connect(
+            database=db_config.database,
+            user=db_config.user,
+            password=db_config.password,
+            host=db_config.host,
+            port=db_config.port
+        )
+        return conn
+
+
+class DataWarehouseDB:
+    """Data Warehouse database connection object for re-use in application."""
+
+    def __init__(self, app=None):
+        """Initialize app context on instantiation."""
+        self.connector = None
+        if app:
+            self.init_app(app)
+
+    def init_app(self, app):
+        """Initialize the app with the Data Warehouse engine and session."""
+        self.connector = Connector(refresh_strategy="lazy")
+
+        db_config = DBConfig(
+            unix_sock=app.config.get("DW_UNIX_SOCKET"),
+            host=app.config.get("DW_HOST"),
+            port=app.config.get("DW_PORT", 5432),
+            database=app.config.get("DW_NAME"),
+            user=app.config.get("DW_USER"),
+            password=app.config.get("DW_PASSWORD"),
+        )
+
+        self.engine = create_engine(
+            "postgresql+pg8000://",
+            creator=lambda: getconn(self.connector, db_config),
+            pool_size=5,
+            max_overflow=2,
+            pool_timeout=10,
+            pool_recycle=1800,
+            connect_args={"use_native_uuid": False}
+        )
+
+        app.teardown_appcontext(self.teardown)
+
+    def teardown(self, exception=None):
+        """Close the connector on teardown."""
+        self.connector.close()
+
+    @property
+    def session(self):
+        """Provide a database session."""
+        return scoped_session(sessionmaker(bind=self.engine))
+
+
+data_warehouse = DataWarehouseDB()