Merge branch 'release-0.2.0' into feat/hamed-remove-kwh-duplicate-1279

.github/workflows/prod-ci.yaml

@@ -42,16 +42,31 @@ jobs:
               echo "IMAGE_TAG retrieved from Test is $imagetag"
               echo "IMAGE_TAG=$imagetag" >> $GITHUB_OUTPUT
 
+  get-current-time:
+    name: Get Current Time
+    runs-on: ubuntu-latest
+    needs: get-image-tag
+
+    outputs:
+      CURRENT_TIME: ${{ steps.get-current-time.outputs.CURRENT_TIME }}
+
+    steps:
+      - id: get-current-time
+        run: |
+          TZ="America/Vancouver"
+          echo "CURRENT_TIME=$(date '+%Y-%m-%d %H:%M:%S %Z')" >> $GITHUB_OUTPUT
+
   # Deplog the image which is running on test to prod
   deploy-on-prod:
 
     name: Deploy LCFS on Prod
     runs-on: ubuntu-latest
-    needs: get-image-tag
+    needs: [get-image-tag, get-current-time]
     timeout-minutes: 60
 
     env:
       IMAGE_TAG: ${{ needs.get-image-tag.outputs.IMAGE_TAG }}
+      CURRENT_TIME: ${{ needs.get-current-time.outputs.CURRENT_TIME }}
 
     steps:
 
@@ -66,9 +81,17 @@ jobs:
         uses: trstringer/[email protected]
         with:
           secret: ${{ github.TOKEN }}
-          approvers: AlexZorkin,kuanfandevops,hamed-valiollahi,airinggov,areyeslo,dhaselhan,Grulin,justin-lepitzki,kevin-hashimoto
+          approvers: AlexZorkin,kuanfandevops,hamed-valiollahi,airinggov,areyeslo,dhaselhan,Grulin
           minimum-approvals: 2
-          issue-title: "LCFS ${{env.IMAGE_TAG }} Prod Deployment"
+          issue-title: "LCFS ${{env.IMAGE_TAG }} Prod Deployment at ${{ env.CURRENT_TIME }}."
+
+      - name: Log in to Openshift
+        uses: redhat-actions/[email protected]
+        with:
+          openshift_server_url: ${{ secrets.OPENSHIFT_SERVER }}
+          openshift_token: ${{ secrets.OPENSHIFT_TOKEN }}
+          insecure_skip_tls_verify: true
+          namespace: ${{ env.PROD_NAMESPACE }}
 
       - name: Tag LCFS images from Test to Prod
         run: |
@@ -88,6 +111,6 @@ jobs:
           git config --global user.name "GitHub Actions"
           git add lcfs/charts/lcfs-frontend/values-prod.yaml
           git add lcfs/charts/lcfs-backend/values-prod.yaml
-          git commit -m "update the version with pre-release number for prod"
+          git commit -m "Update image tag ${{env.IMAGE_TAG }} for prod"
           git push
               

.github/workflows/test-ci.yaml

@@ -225,7 +225,7 @@ jobs:
         uses: trstringer/[email protected]
         with:
           secret: ${{ github.TOKEN }}
-          approvers: AlexZorkin,kuanfandevops,hamed-valiollahi,airinggov,areyeslo,dhaselhan,Grulin,justin-lepitzki,kevin-hashimoto
+          approvers: AlexZorkin,kuanfandevops,hamed-valiollahi,airinggov,areyeslo,dhaselhan,Grulin,kevin-hashimoto
           minimum-approvals: 1
           issue-title: "LCFS ${{ env.VERSION }}-${{ env.PRE_RELEASE }} Test Deployment"
 

backend/lcfs/services/keycloak/authentication.py

@@ -2,9 +2,8 @@
 
 import httpx
 import jwt
-from fastapi import HTTPException, Depends
-from redis import ConnectionPool
-from redis.asyncio import Redis
+from fastapi import HTTPException
+from redis.asyncio import Redis, ConnectionPool
 from sqlalchemy import func
 from sqlalchemy.exc import NoResultFound
 from sqlalchemy.ext.asyncio import async_sessionmaker
@@ -27,7 +26,7 @@ class UserAuthentication(AuthenticationBackend):
 
     def __init__(
         self,
-        redis_pool: Redis,
+        redis_pool: ConnectionPool,
         session_factory: async_sessionmaker,
         settings: Settings,
     ):
@@ -39,30 +38,46 @@ def __init__(
         self.test_keycloak_user = None
 
     async def refresh_jwk(self):
-        # Try to get the JWKS data from Redis cache
-        jwks_data = await self.redis_pool.get("jwks_data")
-
-        if jwks_data:
-            jwks_data = json.loads(jwks_data)
-            self.jwks = jwks_data.get("jwks")
-            self.jwks_uri = jwks_data.get("jwks_uri")
-            return
-
-        # If not in cache, retrieve from the well-known endpoint
-        async with httpx.AsyncClient() as client:
-            oidc_response = await client.get(self.settings.well_known_endpoint)
-            jwks_uri = oidc_response.json().get("jwks_uri")
-            certs_response = await client.get(jwks_uri)
-            jwks = certs_response.json()
-
-        # Composite object containing both JWKS and JWKS URI
-        jwks_data = {"jwks": jwks, "jwks_uri": jwks_uri}
-
-        # Cache the composite JWKS data with a TTL of 1 day (86400 seconds)
-        await self.redis_pool.set("jwks_data", json.dumps(jwks_data), ex=86400)
-
-        self.jwks = jwks
-        self.jwks_uri = jwks_uri
+        """
+        Refreshes the JSON Web Key (JWK) used for token verification.
+        This method attempts to retrieve the JWK from Redis cache.
+        If not found, it fetches it from the well-known endpoint
+        and stores it in Redis for future use.
+        """
+        # Create a Redis client from the connection pool
+        async with Redis(connection_pool=self.redis_pool) as redis:
+            # Try to get the JWKS data from Redis cache
+            jwks_data = await redis.get("jwks_data")
+
+            if jwks_data:
+                jwks_data = json.loads(jwks_data)
+                self.jwks = jwks_data.get("jwks")
+                self.jwks_uri = jwks_data.get("jwks_uri")
+                return
+
+            # If not in cache, retrieve from the well-known endpoint
+            async with httpx.AsyncClient() as client:
+                oidc_response = await client.get(self.settings.well_known_endpoint)
+                oidc_response.raise_for_status()
+                jwks_uri = oidc_response.json().get("jwks_uri")
+
+                if not jwks_uri:
+                    raise ValueError(
+                        "JWKS URI not found in the well-known endpoint response."
+                    )
+
+                certs_response = await client.get(jwks_uri)
+                certs_response.raise_for_status()
+                jwks = certs_response.json()
+
+            # Composite object containing both JWKS and JWKS URI
+            jwks_data = {"jwks": jwks, "jwks_uri": jwks_uri}
+
+            # Cache the composite JWKS data with a TTL of 1 day (86400 seconds)
+            await redis.set("jwks_data", json.dumps(jwks_data), ex=86400)
+
+            self.jwks = jwks
+            self.jwks_uri = jwks_uri
 
     async def authenticate(self, request):
         # Extract the authorization header from the request

backend/lcfs/services/rabbitmq/transaction_consumer.py

@@ -4,7 +4,7 @@
 
 from redis.asyncio import Redis
 from sqlalchemy.ext.asyncio import AsyncSession
-from lcfs.dependencies.dependencies import get_redis_pool
+from lcfs.services.redis.dependency import get_redis_pool
 from fastapi import Request
 
 from lcfs.db.dependencies import async_engine
@@ -50,14 +50,14 @@ async def process_message(self, body: bytes, request: Request):
         compliance_units = message_content.get("compliance_units_amount")
         org_id = message_content.get("organization_id")
 
-        redis = await get_redis_pool(request)
+        redis_pool = await get_redis_pool(request)
 
         async with AsyncSession(async_engine) as session:
             async with session.begin():
                 repo = OrganizationsRepository(db=session)
                 transaction_repo = TransactionRepository(db=session)
                 redis_balance_service = RedisBalanceService(
-                    transaction_repo=transaction_repo, redis_pool=redis.connection_pool
+                    transaction_repo=transaction_repo, redis_pool=redis_pool
                 )
                 org_service = OrganizationsService(
                     repo=repo,

backend/lcfs/services/redis/dependency.py

@@ -1,26 +1,23 @@
-from typing import AsyncGenerator
-
-from redis.asyncio import Redis
+from redis.asyncio import ConnectionPool
 from starlette.requests import Request
 
 
+# Redis Pool Dependency
 async def get_redis_pool(
     request: Request,
-) -> AsyncGenerator[Redis, None]:  # pragma: no cover
+) -> ConnectionPool:
     """
-    Returns connection pool.
-
-    You can use it like this:
-
-    >>> from redis.asyncio import ConnectionPool, Redis
-    >>>
-    >>> async def handler(redis_pool: ConnectionPool = Depends(get_redis_pool)):
-    >>>     async with Redis(connection_pool=redis_pool) as redis:
-    >>>         await redis.get('key')
+    Returns the Redis connection pool.
 
-    I use pools, so you don't acquire connection till the end of the handler.
+    Usage:
+        >>> from redis.asyncio import ConnectionPool, Redis
+        >>>
+        >>> async def handler(redis_pool: ConnectionPool = Depends(get_redis_pool)):
+        >>>     redis = Redis(connection_pool=redis_pool)
+        >>>     await redis.get('key')
+        >>>     await redis.close()
 
-    :param request: current request.
-    :returns:  redis connection pool.
+    :param request: Current request object.
+    :returns: Redis connection pool.
     """
     return request.app.state.redis_pool

backend/lcfs/services/redis/lifetime.py

@@ -1,26 +1,30 @@
 import logging
 from fastapi import FastAPI
-from redis import asyncio as aioredis
+from redis.asyncio import ConnectionPool, Redis
 from redis.exceptions import RedisError
 
 from lcfs.settings import settings
 
 logger = logging.getLogger(__name__)
 
+
 async def init_redis(app: FastAPI) -> None:
     """
     Creates connection pool for redis.
 
     :param app: current fastapi application.
     """
     try:
-        app.state.redis_pool = aioredis.from_url(
+        app.state.redis_pool = ConnectionPool.from_url(
             str(settings.redis_url),
             encoding="utf8",
             decode_responses=True,
-            max_connections=200
+            max_connections=200,
         )
-        await app.state.redis_pool.ping()
+        # Test the connection
+        redis = Redis(connection_pool=app.state.redis_pool)
+        await redis.ping()
+        await redis.close()
         logger.info("Redis pool initialized successfully.")
     except RedisError as e:
         logger.error(f"Redis error during initialization: {e}")
@@ -29,6 +33,7 @@ async def init_redis(app: FastAPI) -> None:
         logger.error(f"Unexpected error during Redis initialization: {e}")
         raise
 
+
 async def shutdown_redis(app: FastAPI) -> None:  # pragma: no cover
     """
     Closes redis connection pool.
@@ -37,8 +42,7 @@ async def shutdown_redis(app: FastAPI) -> None:  # pragma: no cover
     """
     try:
         if hasattr(app.state, "redis_pool"):
-            await app.state.redis_pool.close()
-            await app.state.redis_pool.wait_closed()
+            await app.state.redis_pool.disconnect(inuse_connections=True)
         logger.info("Redis pool closed successfully.")
     except RedisError as e:
         logger.error(f"Redis error during shutdown: {e}")

backend/lcfs/services/s3/client.py

@@ -7,7 +7,7 @@
 from sqlalchemy import select
 from sqlalchemy.exc import InvalidRequestError
 from sqlalchemy.ext.asyncio import AsyncSession
-from lcfs.dependencies.dependencies import get_s3_client
+from lcfs.services.s3.dependency import get_s3_client
 
 from lcfs.db.dependencies import get_async_db_session
 from lcfs.db.models.compliance import ComplianceReport
@@ -28,13 +28,13 @@
 class DocumentService:
     def __init__(
         self,
-        request: Request,
         db: AsyncSession = Depends(get_async_db_session),
         clamav_service: ClamAVService = Depends(),
+        s3_client=Depends(get_s3_client),
     ):
         self.db = db
         self.clamav_service = clamav_service
-        self.s3_client = request.app.state.s3_client
+        self.s3_client = s3_client
 
     @repo_handler
     async def upload_file(self, file, parent_id: str, parent_type="compliance_report"):

backend/lcfs/services/s3/dependency.py

@@ -0,0 +1,19 @@
+from starlette.requests import Request
+import boto3
+
+
+# S3 Client Dependency
+async def get_s3_client(
+    request: Request,
+) -> boto3.client:
+    """
+    Returns the S3 client from the application state.
+
+    Usage:
+        >>> async def handler(s3_client = Depends(get_s3_client)):
+        >>>     s3_client.upload_file('file.txt', 'my-bucket', 'file.txt')
+
+    :param request: Current request object.
+    :returns: S3 client.
+    """
+    return request.app.state.s3_client

backend/lcfs/services/s3/lifetime.py

@@ -0,0 +1,30 @@
+import boto3
+from fastapi import FastAPI
+from lcfs.settings import settings
+
+
+async def init_s3(app: FastAPI) -> None:
+    """
+    Initialize the S3 client and store it in the app state.
+
+    :param app: FastAPI application.
+    """
+    app.state.s3_client = boto3.client(
+        "s3",
+        aws_access_key_id=settings.s3_access_key,
+        aws_secret_access_key=settings.s3_secret_key,
+        endpoint_url=settings.s3_endpoint,
+        region_name="us-east-1",
+    )
+    print("S3 client initialized.")
+
+
+async def shutdown_s3(app: FastAPI) -> None:
+    """
+    Cleanup the S3 client from the app state.
+
+    :param app: FastAPI application.
+    """
+    if hasattr(app.state, "s3_client"):
+        del app.state.s3_client
+        print("S3 client shutdown.")