# /factory/registry_factory.py
class RegistryFactory:
"""Get the appropriate summary computation class based on the registrant type"""
@staticmethod
def get_registry_class(
target_registry,
) -> RegistryInterface:
if target_registry == G2PRegistryType.FARMER.value:
return RegistryFarmer()
# add multiple interfaces using elif blocks
else:
raise BGTaskException(code=BGTaskErrorCodes.INVALID_REQUEST)
# /schema/registry_farmer.py
class G2PFarmerRegistryPayload(G2PRegistryPayload):
## ... add fields from registry view defined ...
# /models/registry_farmer.py
class G2PFarmerRegistry(G2PRegistry):
__tablename__ = "g2p_farmer_registry" # table name in registry view
## ... add fields from registry view defined ...
| Method Name | Type | Purpose | Key Arguments | Returns | Implementation Notes |
|---|---|---|---|---|---|
get_summary | Async | Retrieves summary statistics for a given beneficiary list asynchronously. | beneficiary_list_id: str, bg_task_session: AsyncSession, formated: bool | BeneficiaryListSummaryPayload | Used in API calls; fetches formatted summary metrics from summary table. |
get_summary_sync | Sync | Same as get_summary but executed synchronously (for Celery or background tasks). | beneficiary_list_id: str, bg_task_session: Session | BeneficiaryListSummaryPayload | Ideal for heavy computation where async isn’t needed. |
compute_eligibility_statistics | Sync | Computes eligibility-based summary metrics for beneficiaries. | beneficiary_list_details: List[BeneficiaryListDetails], base_summary, sr_session, bg_task_session | None | Uses NumPy for percentile and mean computations; updates summary model. |
compute_entitlement_statistics | Sync | Computes entitlement statistics (e.g., payment distribution by gender). | beneficiary_list_id: str, bg_task_session: Session, sr_session: Session | None | Groups entitlements by benefit_code_id; calculates mean, Q1, Q2, Q3. |
get_registrants_by_ids | Sync | Fetches registrant data from the registry database. | registrant_ids: List[str], sr_session: Session | List[G2PRegistry] | Uses chunked loading (yield_per(500)) for performance on large datasets. |
get_is_registant_entitled | Sync | Checks if a registrant satisfies entitlement criteria using a SQL query. | registrant_id: str, sql_query: str, sr_session: Session | bool | Constructs validated dynamic SQL using construct_get_is_registrant_entitled_sql_query. |
get_entitlement_multiplier | Sync | Retrieves multiplier value for entitlement scaling. | multiplier: str, registrant_id: str, sr_session: Session | int | Executes a SQL query; defaults to 1 if not found or multiplier is "none". |
search_beneficiaries | Async | Performs paginated and filtered beneficiary searches. | bg_task_session: AsyncSession, sr_session: AsyncSession, beneficiary_list_id: str, target_registry: str, search_query, page, page_size, order_by | BeneficiarySearchResponsePayload | Builds dynamic SQL queries with construct_beneficiary_search_sql_query and applies caching. |
construct_multiplier_sql_query | Utility | Builds SQL query to fetch multiplier column from registry table. | multiplier: str, target_registry: str | TextClause | Returns a prepared SQLAlchemy text() object. |
construct_beneficiary_search_sql_query | Utility | Constructs SQL for paginated search with WHERE and ORDER BY. | registrant_ids: List[str], target_registry: str, where_clause: str, order_by: str, page_size: int, page: int | (TextClause, Dict[str, Any]) | Replaces curly quotes in filters; dynamically injects pagination params. |
construct_beneficiary_search_count_sql_query | Utility | Builds SQL query to count total search results. | registrant_ids: List[str], target_registry: str, where_clause: str | (TextClause, Dict[str, Any]) | Mirrors main query but replaces SELECT * with SELECT COUNT(*). |
construct_get_is_registrant_entitled_sql_query | Utility | Prepares validated entitlement SQL query with a dynamic WHERE clause. | registrant_id: str, target_registry: str, sql_query: str | TextClause | Validates SQL starts with SELECT; appends correct registry table reference. |