move kafka to threadings and setup database connection

app/__init__.py

@@ -3,6 +3,7 @@ from flask_cors import CORS
 from app.config import Config
 from app.routes import auth_bp, autocall_bp
 from app.errors import method_not_allowed, unsupported_media_type
+from app.extensions import db
 
 
 def create_app():
@@ -23,4 +24,7 @@ def create_app():
     app.register_error_handler(405, method_not_allowed)
     app.register_error_handler(415, unsupported_media_type)
 
+    # Database
+    db.init_app(app)
+
     return app

app/config.py

@@ -28,5 +28,15 @@ class Config:
         "BANK_CALL_BASIC_AUTH_PASSWORD", "password"
     )
 
+    DATABASE_USER = os.getenv("DATABASE_USER")
+    DATABASE_PASSWORD = os.getenv("DATABASE_PASSWORD")
+    DATABASE_HOST = os.getenv("DATABASE_HOST")
+    DATABASE_NAME = os.getenv("DATABASE_NAME")
+    DATABASE_PORT = os.getenv("DATABASE_PORT", 10532)
+
+    SQLALCHEMY_DATABASE_URI = f"postgresql+psycopg2://{DATABASE_USER}:{DATABASE_PASSWORD}@{DATABASE_HOST}:{DATABASE_PORT}/{DATABASE_NAME}"
+    SQLALCHEMY_TRACK_MODIFICATIONS = False
+
+
 
 settings = Config()

app/extensions.py

@@ -0,0 +1,3 @@
+from flask_sqlalchemy import SQLAlchemy
+
+db = SQLAlchemy()

openapi.yml

@@ -98,3 +98,21 @@ paths:
       responses:
         200:
           description: A successful response
+  /autocall/refresh-verify-disbursement:
+    get:
+      summary: Refresh the disbursement to verify
+      responses:
+        200:
+          description: A successful response
+  /autocall/refresh-disbursement:
+    get:
+      summary: Refresh the disbursement
+      responses:
+        200:
+          description: A successful response
+  /autocall/payment-callback:
+    get:
+      summary: The Payment callback
+      responses:
+        200:
+          description: A successful response

requirements.txt

@@ -6,3 +6,6 @@ Flask-Cors==3.0.10
 gunicorn
 requests
 confluent-kafka==1.9.2
+flask-sqlalchemy
+psycopg2-binary
+alembic

wsgi.py

@@ -1,17 +1,13 @@
+import threading
 from app import create_app
 from app.integrations import KafkaIntegration
 from app.config import settings
 from app.utils.logger import logger
 
 app = create_app()
+kafka = KafkaIntegration()
 
-if __name__ != "__main__":
+def start_kafka_consumer():
-
-    #Expose WSGI app instance for Gunicorn
-    wsgi_app = app
-
-    kafka = KafkaIntegration()
-
     logger.info("Starting Kafka consumer...")
     while True:
         try:
@@ -28,8 +24,12 @@ if __name__ != "__main__":
 
         except Exception as e:
             logger.error(f"Error while receiving message: {e}")
-            raise
 
 
+if __name__ != "__main__":
+
     # Expose WSGI app instance for Gunicorn
-    # wsgi_app = app
+    wsgi_app = app
+
+    # Start kafka in a thread
+    # threading.Thread(target=start_kafka_consumer, daemon=True).start()