Key Components

info

This page provides a detailed overview of the major components in AquaGen API and how they work together.

🧩 Component Overview

1️⃣ Flask Application (app/__init__.py)

The central Flask application that bootstraps the entire system.

Key Responsibilities

• Initialize Flask app with CORS and JWT
• Register API blueprints (user, admin, external, qa)
• Configure Azure Application Insights telemetry
• Set up custom template filters
• Configure JWT token lifecycle
• Implement user context loading

Configuration

# JWT Configuration
app.config["JWT_TOKEN_LOCATION"] = ["headers", "query_string"]
app.config["JWT_SECRET_KEY"] = Config.SECRET_KEY
app.config["JWT_ACCESS_TOKEN_EXPIRES"] = timedelta(hours=4)
app.config["JWT_REFRESH_TOKEN_EXPIRES"] = timedelta(days=14)

Custom Template Filters

Filter	Purpose	Example Output
decimal_format	Format numbers with precision	123.456 → 123.46
date_time_to_date	Convert UTC datetime to date	2024-11-09T10:30:00Z → 09/11/2024
date_to_date	Format date strings	2024-11-09 → 09/11/2024
date_time_to_time	Extract time from datetime	2024-11-09T10:30:00Z → 10:30:00

Middleware Stack

2️⃣ API Blueprints (app/apis/)

User Blueprint (user.py)

Purpose: User-facing API endpoints

Namespaces (35 total):

• user - User profile and settings
• accounts - Account management
• reports - Report generation
• device_data - Device data retrieval
• device_data_v2 - Enhanced device data API
• alerts - Alert management
• notifications - Notification endpoints
• landing_page - Dashboard data
• summary_page - Summary reports
• water_balance - Water balance analysis
• granular_data - Time-series data
• quality_data - Quality parameters
• And more...

Authentication: All endpoints require JWT (@jwt_required())

Admin Blueprint (admin.py)

Purpose: Administrative operations

Namespaces (24 total):

• admin_login - Admin authentication
• industry - Industry CRUD operations
• unit - Unit/device management
• user - User management
• unit_data - Admin unit data access
• workspace - Workspace configuration
• automated_report - Scheduled reports
• offline_devices - Offline device tracking
• notification - System notifications
• insights - Analytics insights
• archive_data - Data migration/archiving
• gchats - Google Chat integration
• validator - Data validation tools
• hourly_consumption_alert_email - Hourly consumption alert emails
• global_logs - Audit logs
• shift_monitoring - Shift tracking
• bp_trigger - Batch processing triggers
• bathy_data_processor - Bathymetry data processing
• plc_event_hub_test - PLC event hub testing
• water_quality_satellite - Satellite water quality (Lake Pulse)
• verify_monthly_data - Monthly data verification
• water_balance_leakage_detection - Leakage detection
• water_balance_leakage_evaluator - Leakage evaluation

Authentication: Requires admin role (@admin_required())

External Blueprint (external.py)

Purpose: Third-party integrations

Namespaces:

• external_login - External system authentication
• water_ratio - Water measurement standards

Blueprint Registration

app.register_blueprint(user_blueprint)   # /api/user
app.register_blueprint(admin_blueprint)  # /api/admin
app.register_blueprint(external_blueprint) # /api/external
app.register_blueprint(qa_blueprint)     # /api/qa

3️⃣ Route Handlers (app/routes/)

info

Route handlers parse HTTP requests, invoke the appropriate service, and pass the result to formatters before returning the HTTP response.

Pre-Route Processing

1. Rate Limiter

flask-limiter applies before any route logic runs.

• Default limits: 200 requests/day, 50 requests/hour per IP
• Key function: X-Forwarded-For header → falls back to remote_addr
• Storage: In-memory (fixed-window strategy)
• Custom limits for specific operations:

Operation	Limits
OTP generate	3/min, 10/hr, 20/day
OTP verify	5/min, 15/hr
Login	10/min, 50/hr
User create	5/hr, 20/day

2. Auth Interceptor (app/auth/auth.py)

Runs as @app.before_request on every request.

Public routes (skip auth entirely):

• /api/user/user/login, /logout, /otplogin, /refresh, /swaggerLogin, /validateCredentials
• /api/admin/auth/login, /refresh, /swaggerLogin
• /api/external/auth/swaggerLogin, /api/qa/auth/swaggerLogin
• All Swagger/docs UI paths (/api/*/docs, /api/*/swagger.json, /static, /swaggerui)
• /api/user/virtual/*, /api/admin/shift-monitoring/*, /api/admin/bpTrigger, /api/admin/bathyDataProcessor/*

For all other routes, the interceptor:

1. Extracts token from Authorization: Bearer <token> header or ?jwt= query param
2. Decodes token to get userId and industryId
3. Calls TokenService.validate_token() — checks token is active in token_logs_container (skipped for INTERNAL industry)
4. Calls verify_jwt_in_request() to verify signature and expiry
5. Sets g.user_id, g.industry_id, g.user_permissions on Flask's request context

Returns 401 for missing/invalid tokens, 440 for session-expired tokens.

3. JWT User Loader

After token verification, Flask-JWT-Extended calls user_lookup_loader to hydrate the current user:

• Loads industry data from industries_container
• Sets g.timezone for the request
• Loads alertsConfig and evaluates leakage monitoring state
• Supports targetIndustryId header for internal/company cross-industry access

4. Timezone Handler

Sets g.timezone = 'Asia/Kolkata' as default if not already set by the JWT user loader (e.g. for public routes).

5. IP Tracker (app/security_checks/track_user.py)

Runs as @app.before_request after auth:

• Resolves client IP from X-Forwarded-For (first entry) or remote_addr
• Skips geo-lookup for private IPs (RFC1918 ranges)
• For public IPs: fetches city, region, country from ipinfo.io (cached via @lru_cache(maxsize=1000))
• Writes a log entry to the database with userId, industryId, ip, city, region, country, created_on

Route-Level Decorators

These are applied per endpoint inside the route handler:

Decorator	Purpose
@jwt_required()	Validates JWT and makes current_user available
@permission_required(p)	Checks claims["permissions"] list, returns 401 if missing
@jwt_or_redirect(url)	For browser routes — redirects to url if not authenticated via cookie
@validate_values(rules)	Scans string params for SQL injection patterns, raises ValidationError (405)

Post-Response: Security Headers

Applied to every response via @app.after_request:

Header	Value
Content-Security-Policy	frame-ancestors <ALLOWED_FRAME_ANCESTORS>
X-Content-Type-Options	nosniff
X-XSS-Protection	1; mode=block
Referrer-Policy	strict-origin-when-cross-origin

Route Files

User API (/api/user)

File	Area
report.py	Report generation (all types and formats)
device_data.py	Sensor readings by date range
device_data_v2.py	Enhanced device data with formatter pipeline
alerts_routes.py	Alert listing and creation
granular_data.py	High-frequency time-series data
landing_page.py	Dashboard summary data
summary_page.py	Summary report data
water_balance.py	Water balance analysis
water_neutrality_page.py	Water Neutrality Index data
ground_water_level_graph_data.py	Groundwater level graphs
gwi.py	GWI (Groundwater Intelligence)
fwi_route.py	FWI (Fresh Water Intelligence)
rainfall_data.py	Rainfall measurements
virtual_device_data.py	Manual / virtual node data entry
notification.py	User notifications
user.py	User profile, login, and settings
accounts.py	Account management
scada.py	SCADA diagram editor
ro_efficiency.py	RO plant efficiency tracking
standard_category_view.py	Standard category metadata
additional_feature.py	Feature flags per industry
subscription.py	Subscription management
plc_register.py	PLC register read/write
send_mail.py	Direct email sending
common.py	Shared utility endpoints
uwms.py	UWMS data
quality_reports_data_extraction.py	Quality report data extraction
batch_processing.py	Batch processing triggers
external_device_data.py	External device data access
executive_summary/executiveSummaryData.py	Executive summary dashboard
executive_summary/executiveSummaryVirtualLogin.py	Executive summary virtual login
gpt/	AquaGPT routes

Admin API (/api/admin)

File	Area
admin/admin_login.py	Admin authentication
admin/industry.py	Industry CRUD and configuration
admin/unit.py	Unit / device management
admin/user.py	User management
admin/device_data.py	Admin device data access
admin/workspace.py	Workspace configuration
admin/automated_report.py	Scheduled report management
admin/offline_devices.py	Offline device tracking
admin/admin_notification.py	System notifications
admin/insights.py	Analytics insights
admin/archive_data.py	Data migration and archiving
admin/gchats.py	Google Chat integration
admin/validator.py	Data validation tools
admin/hourly_consumption_alert_email.py	Hourly consumption alert emails
admin/global_logs.py	Audit logs
admin/shift_monitoring.py	Shift tracking
admin/bp_trigger.py	Batch processing triggers
admin/bathy_data_processor.py	Bathymetry data processing
admin/plc_eventhub.py	PLC event hub testing
admin/water_quality_satellite.py	Satellite water quality (Lake Pulse)
admin/verify_monthly_data.py	Monthly data verification
admin/water_balance_leakage_formula.py	Leakage detection
admin/water_balance_leakage_evaluator.py	Leakage evaluation

External API (/api/external)

File	Area
external/external_login.py	External system authentication
external/water_ratio_data.py	Water measurement standards

4️⃣ Service Layer (app/services/)

info

Services implement all business logic. A route handler instantiates a service with the parsed request data and calls a method on it.

The service:

1. Reads industry/user context from current_user (hydrated by the JWT user loader on every request) — no extra auth calls needed
2. Determines which units, categories, or date ranges are relevant based on request parameters
3. Calls DatabaseSupporter static methods to fetch raw data from Cosmos DB — often in parallel using ThreadPoolExecutor for multi-unit or multi-date queries
4. Applies business logic — aggregations, threshold checks, graph computations, date/timezone conversions
5. Returns structured Python dicts/lists back to the route handler

The route handler then passes this data to a formatter. Services do not call formatters directly.

Key patterns across services:

• current_user is always available — contains industryId, unitsMapping, categories, daysOffset, alertsConfig, and full industry metadata
• Date/time handling goes through DateTimeUtil for timezone-aware conversions
• Water balance, GWI, FWI, and landing page services load their configuration documents from additional_feature_container at init time
• Alert processing resolves recipient contacts from users_container, then dispatches to channel classes (Email, SMS, WhatsApp, FCM, Web) based on per-recipient channel config

Report Services

ReportService (Dispatcher)

app/services/report/report.py

The main dispatcher. Accepts a ReportRequestData object and routes to the appropriate report class based on serviceType.

ServiceType	Class	Description
water	FlowReport	Water flow/consumption reports
energy	EnergyReport	Energy consumption reports
level	LevelReport	Water level/tank reports
quality	QualityReport	Water quality parameter reports
borewell	BorewellReport	Groundwater borewell reports
rain_water	RainWaterReport	Rainfall collection reports
summary	SummaryReport	Multi-category summary reports
smart_city	SmartCityReport	Smart city monitoring reports
water_balance	WaterBalanceReport	Inflow vs outflow analysis
alert	AlertReport	Alert history reports
consolidated	ConsolidatedReport	Monthly consolidated reports
custom_water	CustomFlowReport	Custom water configuration
custom_energy	CustomEnergyReport	Custom energy configuration
custom_level	CustomLevelReport	Custom level configuration
custom_quality	CustomQualityReport	Custom quality configuration
custom_withdrawal	CustomWithdrawalReport	Custom withdrawal reports
ground_water_withdrawal	GroundWaterWithdrawalReport	GW extraction reports
executive	ExecutiveReport	Executive summary reports

Each report class in app/services/report/report_types/ follows this pattern:

1. Accepts ReportRequestData
2. Queries DatabaseSupporter for raw device data
3. Aggregates and calculates metrics (daily totals, hourly breakdowns, etc.)
4. Returns structured data dict consumed by ReportFormatter

Alert Services

AlertsProcessor

app/services/alerts/alerts_processor.py

The central alert evaluation engine. For each alert trigger:

1. Evaluates alert condition against current data (threshold, offline, flow anomaly, quality)
2. Resolves recipient contacts from user IDs
3. Dispatches notifications across enabled channels

AlertsProcessor dispatches to a content class based on alert_type:

Alert Type	Content Class
unit_threshold	UnitThresholdAlertContent
hourly_threshold	HourlyThresholdAlertContent
monthly_threshold	MonthlyThresholdAlertContent
device_status	DeviceStatusAlertContent
level_exceed	LevelAlertContent
stable_flow_alert	StableFlowAlertContent
flow_rate_*	FlowRateAlertContent
quality_threshold	QualityThresholdAlertContent
energy_threshold	EnergyAlertContent
no_flow_n_days	NoFlowAlertContent
no_level_variation_n_days	NoLevelVariationAlertContent
water_balance_leakage	WaterBalanceLeakageAlertContent

Notification channels: EmailChannel (SMTP), FirebaseChannel (FCM push), MessageChannel (SMS), WhatsappChannel, WebChannel (Azure Web PubSub).

Service Groups

User-Facing Services

Service	Purpose
device_data.py	Fetch and process IoT sensor readings by date/category
device_data_v2/	Enhanced device data pipeline with per-category formatters
latest_data_service.py	Most recent reading per unit
granula_data_service.py	High-frequency time-series data
landing_page/	Dashboard aggregations (today, yesterday, past 3 months)
summary_page.py	Summary report data aggregation
water_balance.py	Inflow/outflow modelling
water_neutrality_service.py	Water Neutrality Index computation
gwi_service.py	Groundwater Intelligence calculations
fwi/	Fresh Water Intelligence and rainfall data
groundwater_level_graph/	Groundwater level trend graphs
ro_efficiency_service.py	Reverse-osmosis plant performance tracking
scada.py	SCADA diagram node/edge configuration
virtual_device_data.py	Manual data entry for nodes without physical sensors
alerts/alerts_processor.py	Evaluate alert rules and dispatch multi-channel notifications
notification.py	User notification management
accounts.py	Account and credential management
user.py	User profile, login, OTP, SSO
report/	Report generation — dispatches to service-specific report classes
executive_summary/	Cross-industry executive dashboard

Admin Services (services/admin/)

Service	Purpose
admin_login.py	Admin authentication
industry.py	Industry CRUD, timezone, and shift configuration
unit.py	Unit/device management
user.py	User management and permissions
workspace.py	Workspace configuration
global_logs.py	Audit log retrieval
insights.py	Analytics and insights
gchats.py	Google Chat webhook management
hourly_consumption_alert_email.py	Hourly consumption alert emails
archive_data.py	Data migration and archiving
bp_trigger.py	Batch processing triggers
bathy_data_processor.py	Bathymetry data processing
water_quality_satellite_processor.py	Satellite water quality processing
verify_monthly_data.py	Monthly data verification
automated_report/automated_report.py	Automated report generation and email delivery
plc_eventhub.py	PLC event hub integration

Notification & Communication Services

Service	Purpose
firebase/firebase_suppoter.py	Firebase FCM push notifications
sms/sms_suppoter.py	SMS delivery
send_mail_service.py	Email via Gmail SMTP

Infrastructure Services

Service	Purpose
token_validation.py	JWT token lifecycle — save, validate, revoke
secret_vault_service.py	Azure Key Vault secret fetching
google_verify_token.py	Google SSO token verification
plc_register_service.py	PLC register read/write
web_socket.py	WebSocket support
external/water_ratio.py	Water ratio standards for external API

5️⃣ Formatters (app/formatters/)

info

Formatters are called directly from route handlers after the service returns data. They receive the raw structured data from the service and transform it into the final response — a JSON dict, a rendered HTML string, an in-memory PDF/XLSX/CSV file stream (BytesIO), or a graph-ready structure.

How Formatters Work

1. Route handler receives processed data from the service
2. Route instantiates the appropriate formatter, passing the data and any format/display arguments
3. Formatter reads current_user context (units mapping, industry meta, alerts config) where needed
4. For device data v2: DeviceDataV2FormatterBuilder selects the correct formatter class based on standardCategoryId — flow, stock, energy, quality, or groundwater level — each extending DeviceDataV2FormatterTemplate
5. Each formatter class implements format() returning the final structure
6. For reports: ReportFormatter renders a Jinja2 HTML template first, then converts to PDF via xhtml2pdf (standard) or Playwright (executive summary), to XLSX via xlsxwriter, or to CSV via pandas — all written to a BytesIO stream (no disk writes)
7. The route returns the formatter output directly as the HTTP response

Device Data Formatters

Formatter	Purpose
device_data_formatter.py	Basic device data response (v1)
device_data_v2/device_data_v2_formatter.py	Entry point for v2 device data pipeline
device_data_v2/formatters/device_data_formatter_template.py	Base class — all v2 formatters extend this
device_data_v2/formatters/flow_device_data_formatter.py	Flow / water category
device_data_v2/formatters/stock_device_data_formatter.py	Stock / tank level category
device_data_v2/formatters/energy_device_data_formatter.py	Energy category
device_data_v2/formatters/quality_device_data_formatter.py	Quality parameters
device_data_v2/formatters/gw_device_data_formatter.py	Groundwater level
device_data_v2/formatters/device_data_formatter_builder.py	Selects correct formatter by standardCategoryId
granular_data_formatter.py	High-frequency time-series data
rainfall_data_formatter.py	Rainfall measurements

Report Formatters (formatters/report/)

Formatter	Purpose
report.py	Entry point — dispatches to HTML/PDF/XLSX/CSV output
flow_formatter.py	Water flow report data
level_formatter.py	Level report data
borewell_formatter.py	Borewell report data
quality_formatter.py	Quality report data
water_balance_formatter.py	Water balance report data
smart_city_formatter.py	Smart city report data
rain_water_formatter.py	Rainwater report data
monthly_consolidated_formatter.py	Monthly consolidated report
daily_summary_formatter.py	Daily summary report
report_util.py / report_util_v2.py	Shared report utility methods
xlsx_formatter/	XLSX-specific formatting (alignment, colour, font, image)

Page / Dashboard Formatters

Formatter	Purpose
industry_data_formatter.py	Industry data loaded on every authenticated request
landing_page_formatter.py	Dashboard landing page data
summary_page_formatter.py	Summary page data
water_balance_data_formatter.py	Water balance graphs and metrics
water_neutrality_data_formatter.py	Water Neutrality Index data
gwi_formatter.py	GWI response
fwi_formatter.py	FWI response
scada_data_formatter.py	SCADA diagram nodes and edges
executive_summary_data_formatter.py	Executive summary dashboard
alerts_formatter.py	Alert list response
notification_data_formatter.py	Notification list response
ro_efficiency/ro_efficiency_formatter.py	RO efficiency metrics

6️⃣ Database Layer (app/database/)

info

The database layer is an abstraction between Cosmos DB configuration and the rest of the project. Services and routes never interact with the Cosmos DB SDK directly — they go through this layer.

Files

File	Purpose
database_config.py	Initialises the Cosmos DB client, creates database references, and exposes named container clients used everywhere else
database_supporter.py	122 static methods covering all read/write operations across every container
queires_list.py	94 centralised query string builders — all Cosmos DB SQL queries defined here, validated against SQL injection via @validate_values()

How It Works

• database_config.py connects to Cosmos DB at startup using credentials from Key Vault and exposes container clients (industries_container, users_container, devices_data_container, etc.) as module-level variables
• DatabaseSupporter imports those container clients and provides static methods that services call — e.g. DatabaseSupporter.get_industry_details_by_id(industry_id)
• Queries (in queires_list.py) builds parameterised query strings; DatabaseSupporter passes these to the Cosmos DB SDK

warning

Never build Cosmos DB query strings inline in services or routes. All queries must be defined in queires_list.py and accessed through DatabaseSupporter to ensure consistent SQL injection protection via @validate_values().

Parallel Queries

tip

Where multiple independent queries are needed (e.g. fetching data for multiple units or date ranges), DatabaseSupporter uses ThreadPoolExecutor to run them concurrently and return combined results.

7️⃣ Configuration (app/config.py)

Purpose: Centralized configuration with Azure Key Vault

class Config(object):
    # Database
    COSMOSDBENDPOINT = smi.get(SecretName.COSMOSDBENDPOINT)
    COSMOSDBKEY = smi.get(SecretName.COSMOSDBKEY)

    # Authentication
    AZURE_AD_APP_ID = smi.get(SecretName.AZURE_AD_APP_ID)
    AZURE_AD_ISSUER = smi.get(SecretName.AZURE_AD_ISSUER)

    # Environment
    ENVIRONMENT = os.environ.get('ENVIRONMENT', 'development')

Security Features:

• All secrets fetched from Azure Key Vault
• No hardcoded credentials
• Environment-specific configuration
• Runtime secret rotation support

8️⃣ Utilities (app/util/)

DateTimeUtil

• Timezone conversion (UTC ↔ Local)
• Date parsing and formatting
• Shift-aware time calculations

NumberUtil

• Number truncation
• Decimal precision handling
• SI unit formatting

SIUnitUtil

• Unit conversion (L, KL, ML, GL)
• Power units (W, KW, MW)
• Quality parameter units

CalculationUtil

• Statistical calculations (min, max, avg)
• Trend analysis
• Percentage calculations

🔟 Data Models (app/models/)

Model Types:

Example Model:

from flask_restx import fields

GenericSuccessModel = {
    'message': fields.String(description='Success message'),
    'status': fields.String(default='success'),
    'status_code': fields.Integer(default=200),
    'data': fields.Raw(description='Response data')
}

---

Component Interaction

All components work together following the layered architecture pattern. Routes handle requests, services process business logic, formatters transform data, and the database supporter manages persistence.

Standard Response Envelope

All JSON responses from AquaGen API follow this structure:

{
  "status": "success" | "failed",
  "status_code": 200,
  "message": "Human-readable message",
  "data": { ... }
}

Error Handling

ValidationError (app/security_checks/input_validator.py):
Raised by @validate_values() when SQL injection patterns are detected in string parameters. Caught by @app.errorhandler(ValidationError), returns JSON with HTTP 405.

Rate limit (429):
flask-limiter returns {"message": "Too many requests. Please try again later.", "status": "failed"}.

Report/Alert flows:
Exceptions inside services and formatters propagate up to the route handler and are returned as {"status": "failed", "status_code": 500, "message": "<error>"}.

Industry not found:
@CachedData.jwt.user_lookup_error_loader aborts with 400 "Industry not found" when the JWT's industryId does not exist in industries_container.

📚 Next Steps

• Database Schema - Database schema and queries
• Current User - Request context and industry data