Mpesa integration initial progress

Backend/app/__init__.py

@@ -10,6 +10,7 @@
 
 from app.config import AppConfig
 from flask_session import Session
+from app.profile.mpesa_client import MpesaClient
 
 db = SQLAlchemy()
 migrate = Migrate()
@@ -53,6 +54,13 @@ def create_app(config_class=AppConfig):
     app.config['JWT_ERROR_MESSAGE_KEY'] = 'msg'
     app.config['JWT_IDENTITY_CLAIM'] = 'sub'
 
+    global mpesa_client
+    mpesa_client = MpesaClient(
+        consumer_key=app.config['MPESA_CONSUMER_KEY'],
+        consumer_secret=app.config['MPESA_CONSUMER_SECRET'],
+        is_sandbox=app.config['MPESA_IS_SANDBOX']
+        )
+
     # Add a loader to convert the JWT subject to string before verification
     @jwt.decode_key_loader
     def decode_key_loader(jwt_headers, jwt_data):

Backend/app/config.py

@@ -40,6 +40,11 @@ class AppConfig:
     JWT_HEADER_NAME = 'Authorization'
     JWT_ERROR_MESSAGE_KEY = 'msg'
 
+    # M-Pesa API settings
+    MPESA_CONSUMER_KEY = os.getenv("MPESA_CONSUMER_KEY")
+    MPESA_CONSUMER_SECRET = os.getenv("MPESA_CONSUMER_SECRET")
+    MPESA_IS_SANDBOX = os.getenv("MPESA_IS_SANDBOX") == "True"
+
 class TestConfig(AppConfig):
     SQLALCHEMY_DATABASE_URI = "sqlite://"
     TESTING = True

Backend/app/models.py

@@ -1,5 +1,6 @@
 from app import db
 from datetime import datetime, timezone
+import enum
 
 # Association table for User (volunteer) skills
 user_skills = db.Table('user_skills',
@@ -116,6 +117,9 @@ class OrgProfile(db.Model):
         cascade="all, delete-orphan",
     )
 
+    # One-to-one relationship
+    mpesa_config = db.relationship("MpesaConfig", uselist=False, backref="org_profile")
+
     def __repr__(self):
         return f"{self.org_name}"
 
@@ -146,6 +150,7 @@ def serialize(self):
             "social_media_links": [
                 link.serialize() for link in self.social_media_links
             ],
+            "mpesa_config": self.mpesa_config.serialize() if self.mpesa_config else None,
         }
         return org_data
 
@@ -259,3 +264,35 @@ def serialize(self):
             "platform": self.platform,
             "url": self.url,
         }
+
+class PaymentType(enum.Enum):
+    PAYBILL = "PB"
+    SEND_MONEY = "SM"
+
+class MpesaConfig(db.Model):
+    id = db.Column(db.Integer, primary_key=True)
+    org_profile_id = db.Column(db.Integer, db.ForeignKey("org_profile.id"), unique=True, nullable=False)
+
+    merchant_name = db.Column(db.String(255), nullable=False)  # M-PESA merchant name
+    payment_type = db.Column(db.Enum(PaymentType), nullable=False)
+
+    # This will store either the paybill number or phone number depending on payment_type
+    identifier = db.Column(db.String(50), nullable=False)  
+
+    created_at = db.Column(db.DateTime, default=datetime.now(timezone.utc))
+    updated_at = db.Column(db.DateTime, default=datetime.now(timezone.utc))
+
+
+    def __repr__(self):
+        return f"<MpesaConfig(merchant_name='{self.merchant_name}', payment_type='{self.payment_type.value}', identifier='{self.identifier}')>"
+
+    def serialize(self):
+        return {
+            "id": self.id,
+            "org_profile_id": self.org_profile_id,
+            "merchant_name": self.merchant_name,
+            "payment_type": self.payment_type.value,
+            "identifier": self.identifier,
+            "created_at": self.created_at.isoformat(),
+            "updated_at": self.updated_at.isoformat(),
+        }

Backend/app/profile/mpesa_client.py

@@ -0,0 +1,110 @@
+import base64
+import requests
+from datetime import datetime, timedelta
+import logging
+
+class MpesaClient:
+    def __init__(self, consumer_key, consumer_secret, is_sandbox=True):
+        """Initialize the M-Pesa client with credentials."""
+        self.consumer_key = consumer_key
+        self.consumer_secret = consumer_secret
+        self.access_token = None
+        self.token_expiry = None
+        self.base_url = "https://sandbox.safaricom.co.ke" if is_sandbox else "https://api.safaricom.co.ke"
+
+    def _generate_auth_string(self):
+        """Generate the base64 encoded auth string."""
+        auth_string = f"{self.consumer_key}:{self.consumer_secret}"
+        return base64.b64encode(auth_string.encode()).decode('utf-8')
+
+    def get_access_token(self):
+        """
+        Get an access token from the M-Pesa API.
+        Checks if current token is still valid before requesting a new one.
+        """
+        # Check if we have a valid token
+        if self.access_token and self.token_expiry:
+            if datetime.now() < self.token_expiry - timedelta(minutes=1):
+                return self.access_token
+
+        try:
+            url = f"{self.base_url}/oauth/v1/generate"
+
+            headers = {
+                "Authorization": f"Basic {self._generate_auth_string()}"
+            }
+
+            params = {
+                "grant_type": "client_credentials"
+            }
+
+            response = requests.get(url, headers=headers, params=params)
+            response.raise_for_status()
+
+            result = response.json()
+
+            self.access_token = result["access_token"]
+            # Set token expiry time (subtracting 5 minutes for safety margin)
+            self.token_expiry = datetime.now() + timedelta(seconds=int(result["expires_in"])) - timedelta(minutes=5)
+
+            return self.access_token
+
+        except requests.exceptions.RequestException as e:
+            raise Exception(f"Failed to get access token: {str(e)}")
+
+    def generate_qr_code(self, qr_code_data):
+        """
+        Generate QR code using the M-Pesa API.
+        Args:
+            merchant_name: Name of the Company/M-Pesa Merchant
+            amount: Total amount for the transaction
+            trx_code: Transaction Type (PB or SM)
+            cpi: Credit Party Identifier (paybill/phone number)
+            ref_no: Transaction Reference
+            size: QR code image size in pixels (default 300)
+        Returns:
+            QR code image data
+        """
+
+        try:
+            # Get fresh access token if needed
+            access_token = self.get_access_token()
+
+            url = f"{self.base_url}/mpesa/qrcode/v1/generate"
+
+            payload = {
+                "MerchantName": qr_code_data.get('MerchantName'),
+                "RefNo": qr_code_data.get('RefNo'),
+                "Amount": str(qr_code_data.get('Amount')),
+                "TrxCode": qr_code_data.get('TrxCode'),
+                "CPI": qr_code_data.get('CPI'),
+                "Size": qr_code_data.get('Size') if 'size' in qr_code_data else 300
+            }
+
+            headers = {
+                "Authorization": f"Bearer {access_token}",
+                "Content-Type": "application/json"
+            }
+
+            response = requests.post(url, json=payload, headers=headers)
+            response.raise_for_status()
+
+            result = response.json()
+
+            if 'QRCode' in result:
+                return result['QRCode']
+            else:
+                raise Exception("QR code not found in response")
+
+        except requests.exceptions.RequestException as e:
+            raise Exception(f"Failed to generate QR code: {str(e)}")
+
+
+# Usage example:
+# client = MpesaClient(
+#     consumer_key="your_consumer_key",
+#     consumer_secret="your_consumer_secret",
+#     is_sandbox=True  # Set to False for production
+# )
+# 
+# token = client.get_access_token()

Backend/app/profile/routes.py

@@ -2,9 +2,9 @@
 from flask_login import login_required, current_user
 from sqlalchemy.orm.exc import NoResultFound
 from app.models import User, SkillsNeeded
-
-
 from app import db
+from app import mpesa_client
+import logging
 
 from app.models import (
     OrgProfile,
@@ -345,4 +345,34 @@ def get_volunteer_profile():
 @profile.route("/profile/volunteer/skills", methods=["GET"])
 def get_all_skills():
     skills = SkillsNeeded.query.all()
-    return jsonify([skill.serialize() for skill in skills]), 200
+    return jsonify([skill.serialize() for skill in skills]), 200
+
+# In your route handler
+@profile.route("/profile/generate-qr", methods=['POST', 'OPTIONS'])
+def generate_qr():
+
+    # Handle CORS preflight request
+    if request.method == 'OPTIONS':
+        response = jsonify({'status': 'ok'})
+        response.headers.add('Access-Control-Allow-Headers', 'Content-Type')
+        response.headers.add('Access-Control-Allow-Methods', 'POST')
+        return response
+
+    # Get JSON data from request
+    qr_data = request.get_json()
+    if not qr_data:
+        return jsonify({'error': 'No data provided'}), 400
+
+    # Validate required fields
+    required_fields = ['MerchantName', 'RefNo', 'Amount', 'TrxCode', 'CPI', 'Size']
+    for field in required_fields:
+        if field not in qr_data:
+            return jsonify({'error': f'Missing required field: {field}'}), 400
+
+    try:
+        # Return generated QR code
+        qr_code = mpesa_client.generate_qr_code(qr_data)
+        return jsonify({"qr_code": qr_code})
+
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500

Backend/create_db.py

@@ -1,7 +1,7 @@
 import logging
 from sqlalchemy import inspect
 from app import create_app, db
-from app.models import User, OrgProfile, OrgInitiatives, OrgProjects, SkillsNeeded, FocusArea, SocialMediaLink
+from sqlalchemy import text
 
 # Set up logging
 logging.basicConfig(level=logging.INFO)
@@ -13,45 +13,60 @@
 def verify_db_connection():
     try:
         with app.app_context():
-            # Test database connection
-            db.session.execute('SELECT 1')
+            db.session.execute(text('SELECT 1'))
             logger.info("Database connection successful")
             return True
     except Exception as e:
         logger.error(f"Database connection failed: {str(e)}")
         return False
-
 
-def reset_db():
+def verify_table_columns(inspector, table_name, expected_columns):
+    """Helper function to verify table columns"""
+    existing_columns = [col['name'] for col in inspector.get_columns(table_name)]
+    missing_columns = [col for col in expected_columns if col not in existing_columns]
+    return missing_columns
+
+def upgrade_db():
     try:
         with app.app_context():
-            logger.info("Starting database initialization...")
+            logger.info("Starting database upgrade...")
 
-            # Create all tables
-            db.create_all()
-            logger.info("Database tables created successfully.")
+            # Verify connection
+            if not verify_db_connection():
+                raise Exception("Database connection failed")
 
-            # Verify tables using inspector
+            # Get current database state
             inspector = inspect(db.engine)
-            tables = inspector.get_table_names()
-            logger.info(f"Created tables: {', '.join(tables)}")
+            existing_tables = inspector.get_table_names()
+            logger.info(f"Current tables: {', '.join(existing_tables)}")
+
+            # Get all expected tables from models
+            model_tables = {model.__tablename__: model for model in db.Model.__subclasses__()}
+            logger.info(f"Expected tables from models: {', '.join(model_tables.keys())}")
 
-            # Additional verification
-            logger.info("Verifying key tables...")
-            expected_tables = ['user', 'org_profile', 'org_initiatives', 'org_projects', 
-                                'skills_needed', 'focus_area', 'social_media_link']
-            missing_tables = [table for table in expected_tables if table not in tables]
+            # Create missing tables
+            for table_name, model in model_tables.items():
+                if table_name not in existing_tables:
+                    logger.info(f"Creating table {table_name}...")
+                    model.__table__.create(db.engine)
+                    logger.info(f"Table {table_name} created successfully")
+
+            # Verify all tables after creation
+            inspector = inspect(db.engine)
+            final_tables = inspector.get_table_names()
+            missing_tables = [table for table in model_tables.keys() if table not in final_tables]
 
             if missing_tables:
-                logger.warning(f"Missing tables: {', '.join(missing_tables)}")
-            else:
-                logger.info("All expected tables are present.")
+                logger.error(f"Failed to create tables: {', '.join(missing_tables)}")
+                raise Exception("Database upgrade incomplete - missing tables")
 
+            logger.info("Database upgrade completed successfully")
             return True
 
     except Exception as e:
-        logger.error(f"Error initializing database: {str(e)}")
+        logger.error(f"Error upgrading database: {str(e)}")
         raise e
 
 if __name__ == "__main__":
-    reset_db()
+    verify_db_connection()
+    upgrade_db()

Backend/requirements.dev.txt

@@ -12,6 +12,7 @@ Flask-JWT-Extended==4.7.0
 gunicorn==21.2.0
 itsdangerous==2.0.1
 python-dotenv==0.21.1
+requests==2.32.3
 SQLAlchemy==2.0.12
 Werkzeug==2.2.2
 WTForms==3.0.1