Project Structure & Development Guide

This guide explains the codebase organization and how to add new features to AquaGen API.

📁 Directory Structure

aquagenapi/
├── app/                          # Main application
│   ├── __init__.py               # Flask app initialization
│   ├── config.py                 # Configuration with Azure Key Vault
│   ├── app.py                    # Application entry point
│   │
│   ├── apis/                     # API Blueprints
│   │   ├── user.py               # User API (31 namespaces)
│   │   ├── admin.py              # Admin API (13 namespaces)
│   │   ├── external.py           # External integrations
│   │   └── qa.py                 # QA endpoints
│   │
│   ├── routes/                   # Route handlers (33 files)
│   │   ├── report.py             # Report generation
│   │   ├── device_data.py        # Device data endpoints
│   │   ├── device_data_v2.py     # Enhanced device data
│   │   ├── alerts_routes.py      # Alert management
│   │   ├── notification.py       # Notifications
│   │   ├── landing_page.py       # Dashboard data
│   │   ├── water_balance.py      # Water balance
│   │   ├── granular_data.py      # Granular sensor data
│   │   ├── user.py               # User operations
│   │   ├── admin/                # Admin routes (15 files)
│   │   │   ├── admin_login.py
│   │   │   ├── automated_report.py
│   │   │   ├── industry.py
│   │   │   ├── unit.py
│   │   │   ├── user.py
│   │   │   └── ...
│   │   └── external/             # External routes
│   │       ├── external_login.py
│   │       └── water_ratio_data.py
│   │
│   ├── services/                 # Business logic (40+ services)
│   │   ├── report/               # Report services
│   │   │   ├── report.py         # Main dispatcher
│   │   │   └── report_types/     # 18 report implementations
│   │   │       ├── flow_report.py
│   │   │       ├── energy_report.py
│   │   │       ├── level_report.py
│   │   │       ├── quality_report.py
│   │   │       ├── borewell_report.py
│   │   │       ├── alert_report.py
│   │   │       ├── consolidated_report.py
│   │   │       ├── water_balance_report.py
│   │   │       └── ...
│   │   ├── device_data.py
│   │   ├── device_data_v2/       # Enhanced device data
│   │   │   ├── device_data_v2.py
│   │   │   └── fetch_data/
│   │   ├── alerts/               # Alert system
│   │   │   ├── alerts_processor.py
│   │   │   ├── alerts_content/   # Alert templates
│   │   │   ├── channels/         # Email, SMS, Chat
│   │   │   └── constants/
│   │   ├── admin/                # Admin services (15 files)
│   │   │   ├── admin_login.py
│   │   │   ├── industry.py
│   │   │   ├── unit.py
│   │   │   ├── user.py
│   │   │   └── automated_report/
│   │   │       ├── automated_report.py
│   │   │       └── report_processor.py
│   │   ├── gpt/                  # AI integration
│   │   │   ├── gpt.py
│   │   │   ├── gpt_handler.py
│   │   │   ├── gpt_config.py
│   │   │   ├── aquagpt/
│   │   │   ├── templates/
│   │   │   └── utils/
│   │   ├── notification.py
│   │   ├── summary/              # Summary services
│   │   ├── landing_page/         # Dashboard logic
│   │   ├── water_balance.py
│   │   ├── granula_data_service.py
│   │   ├── water_neutrality_service.py
│   │   ├── powerdrill/           # PowerDrill integration
│   │   ├── quality_data_extraction/
│   │   ├── graph_manager/
│   │   ├── info_builders/
│   │   ├── wri/                  # WRI bathymetry
│   │   ├── firebase/
│   │   ├── sms/
│   │   ├── secret_vault_service.py
│   │   └── web_socket.py
│   │
│   ├── formatters/               # Data formatting (52 files)
│   │   ├── report/               # Report formatters
│   │   │   ├── report.py
│   │   │   ├── report_util.py
│   │   │   ├── xlsx_formatter/
│   │   │   ├── flow_formatter.py
│   │   │   ├── energy_formatter.py
│   │   │   ├── level_formatter.py
│   │   │   ├── quality_formatter.py
│   │   │   └── ...
│   │   ├── device_data_formatter.py
│   │   ├── device_data_v2/       # Polymorphic formatters
│   │   │   ├── device_data_v2_formatter.py
│   │   │   └── formatters/
│   │   │       ├── flow_device_data_formatter.py
│   │   │       ├── energy_device_data_formatter.py
│   │   │       ├── quality_device_data_formatter.py
│   │   │       └── ...
│   │   ├── alerts_formatter.py
│   │   ├── common_data_formatter.py
│   │   ├── graph_data_formatter.py
│   │   └── ...
│   │
│   ├── database/                 # Database layer
│   │   ├── database_config.py    # Cosmos DB setup
│   │   ├── database_supporter.py # 200+ query methods
│   │   └── queires_list.py       # Query templates
│   │
│   ├── models/                   # Data models (30 files)
│   │   ├── error_models.py
│   │   ├── success_models.py
│   │   ├── user_models.py
│   │   ├── industry_models.py
│   │   ├── unit_models.py
│   │   ├── device_data_models.py
│   │   ├── alerts_models.py
│   │   ├── report_models.py
│   │   └── ...
│   │
│   ├── data_class/               # Data classes & DTOs
│   │   ├── report.py
│   │   ├── device_data_v2_args.py
│   │   ├── water_ratio.py
│   │   └── alert/
│   │
│   ├── enum/                     # Enums (20 files)
│   │   ├── service_type.py
│   │   ├── report_type.py
│   │   ├── report_format.py
│   │   ├── category_type.py
│   │   ├── response_status.py
│   │   └── ...
│   │
│   ├── templates/                # Jinja2 templates (55 files)
│   │   ├── common/
│   │   ├── flow/                 # Water reports (7 templates)
│   │   ├── energy/               # Energy reports (6 templates)
│   │   ├── level/                # Level reports (3 templates)
│   │   ├── quality/              # Quality reports (4 templates)
│   │   ├── borewell/             # Borewell reports (4 templates)
│   │   ├── consolidated/
│   │   ├── summary/
│   │   ├── alert/
│   │   └── ...
│   │
│   ├── validators/               # Input validation
│   │   ├── schema.py
│   │   ├── validator_util.py
│   │   └── schemas/              # JSON schemas
│   │
│   ├── security_checks/          # Security layer
│   │   ├── input_validator.py    # SQL injection prevention
│   │   └── track_user.py         # IP tracking
│   │
│   ├── util/                     # Utilities (10 files)
│   │   ├── date_time_util.py
│   │   ├── number_util.py
│   │   ├── calculation_util.py
│   │   ├── si_unit_util.py
│   │   ├── sort_util.py
│   │   └── ...
│   │
│   ├── auth/                     # Authentication
│   │   └── auth.py
│   │
│   ├── constants/                # App constants
│   │   ├── constants.py
│   │   └── templates.py
│   │
│   ├── cachedData/               # In-memory cache
│   │   └── cachedData.py
│   │
│   ├── data/                     # Static data files
│   │   ├── bathyData.py
│   │   └── radhakunj_boundary.geojson
│   │
│   ├── urls/                     # URL configuration
│   │
│   └── logging_config.py
│
├── scripts/                      # Automation scripts
│   └── auto_reports.sh
│
├── docs/                         # Documentation
│   ├── intro.md
│   ├── getting-started.md
│   ├── architecture/
│   ├── core-concepts/
│   ├── api-reference/
│   ├── guides/
│   ├── deployment/
│   ├── advanced/
│   └── development/
│
├── data/                         # Data files
│
├── requirements.txt              # Dependencies
├── app.py                        # Entry point
├── CLAUDE.md                     # Development guidelines
└── README.md

🏗️ Architecture Layers

🎯 Adding New Features

1. Adding a New Report Type

Step 1: Create Report Service

# app/services/report/report_types/my_custom_report.py

from app.services.report.report_types.base_report import BaseReport

class MyCustomReport(BaseReport):
    def get_report(self):
        # Fetch data
        data = self.fetch_data()

        # Process and aggregate
        processed = self.process_data(data)

        # Calculate metrics
        metrics = self.calculate_metrics(processed)

        return {
            'data': processed,
            'metrics': metrics,
            'fileName': self.generate_filename()
        }

    def fetch_data(self):
        # Query database
        from app.database.database_supporter import DatabaseSupporter
        return DatabaseSupporter.get_device_data_by_date_range(
            self.report_request.industry_id,
            self.report_request.start_date,
            self.report_request.end_date
        )

    def process_data(self, data):
        # Aggregate logic
        pass

    def calculate_metrics(self, data):
        # Calculate min, max, avg, etc.
        pass

Step 2: Register in ReportService

# app/services/report/report.py

from app.services.report.report_types.my_custom_report import MyCustomReport

class ReportService:
    def get_report(self):
        if self.report_request.serviceType == ServiceType.MY_CUSTOM:
            return MyCustomReport(self.report_request).get_report()
        # ... other report types

Step 3: Add ServiceType Enum

# app/enum/service_type.py

class ServiceType(Enum):
    # ... existing types
    MY_CUSTOM = 'my_custom'

Step 4: Create Template

<!-- app/templates/my_custom/my_custom_daily_report.html -->

<!DOCTYPE html>
<html>
<head>
    <title>{{ industry_name }} - Custom Report</title>
</head>
<body>
    <h1>Custom Report</h1>
    <table>
        {% for row in data %}
        <tr>
            <td>{{ row.timestamp | date_time_to_date }}</td>
            <td>{{ row.value | decimal_format(2) }}</td>
        </tr>
        {% endfor %}
    </table>
</body>
</html>

Step 5: Create Formatter (if needed)

# app/formatters/report/my_custom_formatter.py

class MyCustomFormatter:
    def format_data(self, data):
        # Format data for template
        pass

2. Adding a New API Endpoint

Step 1: Create Route Handler

# app/routes/my_feature.py

from flask import request
from flask_jwt_extended import jwt_required, current_user
from flask_restx import Resource, Namespace

from app.services.my_feature_service import MyFeatureService
from app.models.success_models import GenericSuccessModel
from app.models.error_models import GenericErrorModel

myFeatureNamespace = Namespace('My Feature',
                                description='My feature APIs',
                                path='/myFeature')

@myFeatureNamespace.route('')
class MyFeatureRoute(Resource):
    @jwt_required()
    @myFeatureNamespace.response(200, 'Success', model=GenericSuccessModel)
    @myFeatureNamespace.response(1401, 'Data not found', model=GenericErrorModel)
    def get(self):
        # Parse parameters
        args = request.args

        # Call service
        service = MyFeatureService(current_user['industryId'])
        result = service.process()

        return {
            'status': 'success',
            'status_code': 200,
            'data': result
        }

Step 2: Create Service

# app/services/my_feature_service.py

from app.database.database_supporter import DatabaseSupporter

class MyFeatureService:
    def __init__(self, industry_id):
        self.industry_id = industry_id

    def process(self):
        # Business logic
        data = DatabaseSupporter.get_data(self.industry_id)
        processed = self.transform(data)
        return processed

    def transform(self, data):
        # Transform data
        pass

Step 3: Register Namespace

# app/apis/user.py

from app.routes.my_feature import myFeatureNamespace

# ... existing code

api.add_namespace(myFeatureNamespace)

3. Adding Database Methods

# app/database/database_supporter.py

class DatabaseSupporter:
    @staticmethod
    def get_my_custom_data(industry_id, filters):
        """
        Get custom data from Cosmos DB
        """
        try:
            container = database_config.my_custom_container

            query = """
                SELECT * FROM c
                WHERE c.industryId = @industry_id
                AND c.type = @type
            """

            parameters = [
                {"name": "@industry_id", "value": industry_id},
                {"name": "@type", "value": filters.get('type')}
            ]

            items = container.query_items(
                query=query,
                parameters=parameters,
                enable_cross_partition_query=True
            )

            return list(items)
        except Exception as e:
            logger.error(f"Error fetching custom data: {e}")
            return []

4. Adding a New Alert Channel

Step 1: Create Channel Handler

# app/services/alerts/channels/my_channel.py

class MyChannelHandler:
    def __init__(self, config):
        self.config = config

    def send(self, alert_data):
        """
        Send alert via my custom channel
        """
        try:
            # Implement channel-specific logic
            self.prepare_message(alert_data)
            self.deliver()
            return True
        except Exception as e:
            logger.error(f"Error sending via MyChannel: {e}")
            return False

    def prepare_message(self, alert_data):
        # Format message
        pass

    def deliver(self):
        # Send message
        pass

Step 2: Register in AlertProcessor

# app/services/alerts/alerts_processor.py

from app.services.alerts.channels.my_channel import MyChannelHandler

class AlertsProcessor:
    def send_notifications(self, alert, channels):
        if 'my_channel' in channels:
            handler = MyChannelHandler(channels['my_channel'])
            handler.send(alert)

🧪 Testing Guidelines

Unit Testing

# tests/test_report_service.py

import unittest
from app.services.report.report import ReportService
from app.data_class.report import ReportRequestData
from app.enum.report_type import ReportType
from app.enum.service_type import ServiceType

class TestReportService(unittest.TestCase):
    def setUp(self):
        self.report_request = ReportRequestData(
            report_type=ReportType.DAILY,
            service_type=ServiceType.WATER,
            industry_id='test-industry',
            start_date='09/11/2024'
        )

    def test_generate_daily_report(self):
        service = ReportService(self.report_request)
        result = service.get_report()

        self.assertIsNotNone(result)
        self.assertIn('data', result)
        self.assertIn('fileName', result)

    def test_invalid_service_type(self):
        self.report_request.service_type = 'invalid'
        with self.assertRaises(ValueError):
            ReportService(self.report_request).get_report()

Integration Testing

# tests/integration/test_report_api.py

import unittest
import requests

class TestReportAPI(unittest.TestCase):
    BASE_URL = 'http://localhost:5001'
    TOKEN = 'your-test-token'

    def test_daily_report_generation(self):
        url = f"{self.BASE_URL}/api/user/report"
        params = {
            'reportType': 'daily',
            'reportFormat': 'html',
            'service': 'water',
            'startDate': '09/11/2024'
        }
        headers = {
            'Authorization': f'Bearer {self.TOKEN}'
        }

        response = requests.get(url, params=params, headers=headers)

        self.assertEqual(response.status_code, 200)
        self.assertIn('<!DOCTYPE html>', response.text)

📝 Code Style Guidelines

Python Style (PEP 8)

# Good
class WaterBalanceService:
    """Service for water balance calculations."""

    def calculate_balance(self, inflow: float, outflow: float) -> dict:
        """
        Calculate water balance.

        Args:
            inflow: Total water inflow in KL
            outflow: Total water outflow in KL

        Returns:
            dict: Balance calculation results
        """
        balance = inflow - outflow
        percentage = (balance / inflow) * 100 if inflow > 0 else 0

        return {
            'balance': balance,
            'percentage': percentage,
            'status': 'surplus' if balance > 0 else 'deficit'
        }

Naming Conventions

Type	Convention	Example
Classes	PascalCase	ReportService, DeviceDataFormatter
Functions	snake_case	get_report(), calculate_metrics()
Variables	snake_case	industry_id, report_data
Constants	UPPER_SNAKE_CASE	MAX_RETRY_COUNT, DEFAULT_TIMEOUT
Files	snake_case	report_service.py, water_balance.py

Import Organization

# Standard library imports
import os
from datetime import datetime, timedelta

# Third-party imports
from flask import request
import pandas as pd

# Local application imports
from app.database.database_supporter import DatabaseSupporter
from app.util.date_time_util import DateTimeUtil
from app.enum.service_type import ServiceType

🔍 Debugging Tips

Enable Debug Mode

# app.py
if __name__ == '__main__':
    app.run(debug=True, port=5001)

Logging

import logging

logger = logging.getLogger(__name__)

# Log levels
logger.debug("Detailed debug information")
logger.info("General information")
logger.warning("Warning message")
logger.error("Error occurred")
logger.critical("Critical error")

Azure Application Insights

from opencensus.ext.azure import metrics_exporter

# Custom metrics
exporter = metrics_exporter.new_metrics_exporter(
    connection_string=Config.APPLICATION_INSIGHTS
)

# Track custom event
from opencensus.stats import aggregation as aggregation_module
from opencensus.stats import measure as measure_module
from opencensus.stats import stats as stats_module
from opencensus.stats import view as view_module

# Define measure
measure = measure_module.MeasureInt("report_generation_time",
                                    "Time to generate report", "ms")

# Track
stats = stats_module.stats
view_manager = stats.view_manager
stats_recorder = stats.stats_recorder

mmap = stats_recorder.new_measurement_map()
tmap = mmap.measure_int_put(measure, 1500)
tmap.record()

---

Development Workflow

1. Create feature branch
2. Implement changes following architecture patterns
3. Write tests
4. Update documentation
5. Submit pull request

📚 Next Steps

• Adding Features Guide - Detailed examples
• Testing Guide - Testing strategies
• API Guidelines - API design principles