airajena commented on code in PR #102:
URL:
https://github.com/apache/fineract-business-intelligence/pull/102#discussion_r4380626079
##########
extractor/extractor.py:
##########
@@ -37,626 +66,508 @@ def as_utc_datetime(value: datetime | None) -> datetime |
None:
return value.astimezone(UTC)
-def _quote_identifier(name: str) -> str:
- return '"' + name.replace('"', '""') + '"'
+def sqlstate(exc: BaseException) -> str | None:
+ payload = exc.args[0] if exc.args else None
+ return payload.get("C") if isinstance(payload, dict) else None
-@dataclass(frozen=True)
-class TableSpec:
- source_table: str
- raw_table: str
- columns: tuple[str, ...]
- primary_key: str
- cursor_column: str = "last_modified_on_utc"
-
-
-TABLE_SPECS: tuple[TableSpec, ...] = (
- TableSpec(
- source_table="m_office",
- raw_table="raw_m_office",
- primary_key="id",
- columns=(
- "id",
- "parent_id",
- "hierarchy",
- "external_id",
- "name",
- "opening_date",
- "created_on_utc",
- "last_modified_on_utc",
- ),
- ),
- TableSpec(
- source_table="m_currency",
- raw_table="raw_m_currency",
- primary_key="id",
- columns=(
- "id",
- "code",
- "decimal_places",
- "currency_multiplesof",
- "display_symbol",
- "name",
- "internationalized_name_code",
- "created_on_utc",
- "last_modified_on_utc",
- ),
- ),
- TableSpec(
- source_table="m_client",
- raw_table="raw_m_client",
- primary_key="id",
- columns=(
- "id",
- "account_no",
- "external_id",
- "status_enum",
- "activation_date",
- "office_joining_date",
- "office_id",
- "staff_id",
- "gender_cv_id",
- "date_of_birth",
- "legal_form_enum",
- "submittedon_date",
- "updated_on",
- "created_on_utc",
- "last_modified_on_utc",
- ),
- ),
- TableSpec(
- source_table="m_product_loan",
- raw_table="raw_m_product_loan",
- primary_key="id",
- columns=(
- "id",
- "short_name",
- "currency_code",
- "currency_digits",
- "currency_multiplesof",
- "principal_amount",
- "min_principal_amount",
- "max_principal_amount",
- "arrearstolerance_amount",
- "name",
- "description",
- "nominal_interest_rate_per_period",
- "annual_nominal_interest_rate",
- "repay_every",
- "repayment_period_frequency_enum",
- "number_of_repayments",
- "overdue_days_for_npa",
- "start_date",
- "close_date",
- "created_on_utc",
- "last_modified_on_utc",
- ),
- ),
- TableSpec(
- source_table="m_loan",
- raw_table="raw_m_loan",
- primary_key="id",
- columns=(
- "id",
- "account_no",
- "external_id",
- "client_id",
- "product_id",
- "loan_status_id",
- "loan_type_enum",
- "currency_code",
- "currency_digits",
- "currency_multiplesof",
- "principal_amount_proposed",
- "principal_amount",
- "approved_principal",
- "net_disbursal_amount",
- "annual_nominal_interest_rate",
- "nominal_interest_rate_per_period",
- "interest_method_enum",
- "interest_calculated_in_period_enum",
- "term_frequency",
- "term_period_frequency_enum",
- "repay_every",
- "repayment_period_frequency_enum",
- "number_of_repayments",
- "amortization_method_enum",
- "submittedon_date",
- "approvedon_date",
- "expected_disbursedon_date",
- "expected_firstrepaymenton_date",
- "disbursedon_date",
- "expected_maturedon_date",
- "maturedon_date",
- "principal_disbursed_derived",
- "principal_repaid_derived",
- "principal_writtenoff_derived",
- "principal_outstanding_derived",
- "interest_charged_derived",
- "interest_repaid_derived",
- "interest_writtenoff_derived",
- "interest_outstanding_derived",
- "fee_charges_outstanding_derived",
- "penalty_charges_outstanding_derived",
- "total_expected_repayment_derived",
- "total_repayment_derived",
- "total_writtenoff_derived",
- "total_outstanding_derived",
- "loan_counter",
- "is_npa",
- "created_on_utc",
- "last_modified_on_utc",
- ),
- ),
- TableSpec(
- source_table="m_loan_transaction",
- raw_table="raw_m_loan_transaction",
- primary_key="id",
- columns=(
- "id",
- "loan_id",
- "office_id",
- "is_reversed",
- "transaction_type_enum",
- "transaction_date",
- "amount",
- "principal_portion_derived",
- "interest_portion_derived",
- "fee_charges_portion_derived",
- "penalty_charges_portion_derived",
- "overpayment_portion_derived",
- "outstanding_loan_balance_derived",
- "submitted_on_date",
- "created_on_utc",
- "last_modified_on_utc",
- ),
- ),
- TableSpec(
- source_table="m_delinquency_range",
- raw_table="raw_m_delinquency_range",
- primary_key="id",
- columns=(
- "id",
- "classification",
- "min_age_days",
- "max_age_days",
- "created_on_utc",
- "last_modified_on_utc",
- ),
- ),
- TableSpec(
- source_table="m_delinquency_bucket",
- raw_table="raw_m_delinquency_bucket",
- primary_key="id",
- columns=(
- "id",
- "name",
- "created_on_utc",
- "last_modified_on_utc",
- ),
- ),
- TableSpec(
- source_table="m_delinquency_bucket_mappings",
- raw_table="raw_m_delinquency_bucket_mappings",
- primary_key="id",
- columns=(
- "id",
- "delinquency_range_id",
- "delinquency_bucket_id",
- "created_on_utc",
- "last_modified_on_utc",
- ),
- ),
- TableSpec(
- source_table="m_loan_delinquency_tag_history",
- raw_table="raw_m_loan_delinquency_tag_history",
- primary_key="id",
- columns=(
- "id",
- "delinquency_range_id",
- "loan_id",
- "addedon_date",
- "liftedon_date",
- "created_on_utc",
- "last_modified_on_utc",
- ),
- ),
- TableSpec(
- source_table="m_loan_repayment_schedule",
- raw_table="raw_m_loan_repayment_schedule",
- primary_key="id",
- cursor_column="lastmodified_date",
- columns=(
- "id",
- "loan_id",
- "fromdate",
- "duedate",
- "installment",
- "principal_amount",
- "principal_completed_derived",
- "principal_writtenoff_derived",
- "interest_amount",
- "interest_completed_derived",
- "interest_writtenoff_derived",
- "interest_waived_derived",
- "fee_charges_amount",
- "fee_charges_completed_derived",
- "fee_charges_writtenoff_derived",
- "fee_charges_waived_derived",
- "penalty_charges_amount",
- "penalty_charges_completed_derived",
- "penalty_charges_writtenoff_derived",
- "penalty_charges_waived_derived",
- "total_paid_in_advance_derived",
- "total_paid_late_derived",
- "completed_derived",
- "obligations_met_on_date",
- "is_re_aged",
- "lastmodified_date",
- ),
- ),
- TableSpec(
- source_table="batch_job_execution",
- raw_table="raw_batch_job_execution",
- primary_key="job_execution_id",
- columns=(
- "job_execution_id",
- "job_name",
- "status",
- "start_time",
- "end_time",
- "exit_code",
- "exit_message",
- "created_on_utc",
- "last_modified_on_utc",
- ),
- ),
-)
+def csv_field(value) -> str:
+ if value is None:
+ return ""
+ if isinstance(value, bool):
+ return "t" if value else "f"
+ if isinstance(value, datetime):
+ text = as_utc_datetime(value).isoformat()
+ elif isinstance(value, (date, Decimal, int, float)):
+ text = str(value)
+ else:
+ text = str(value)
+ return '"' + text.replace('"', '""') + '"'
+
+
+def build_copy_payload(rows: Iterable[Sequence]) -> io.StringIO:
+ buf = io.StringIO()
+ for row in rows:
+ buf.write(",".join(csv_field(v) for v in row))
+ buf.write("\n")
+ buf.seek(0)
+ return buf
-class FineractExtractor:
- def __init__(self, config: AppConfig) -> None:
- self.config = config
+@dataclass(frozen=True)
+class Assignment:
+ table: TableContract
+ target: str
+ epoch: int
+
+
+def plan_assignments(
+ ownerships: Iterable[Ownership], contract: Contract, tables: set[str] |
None = None
+) -> list[Assignment]:
+ plan = []
+ for o in ownerships:
+ if o.cutover_state is CutoverState.PAUSED or (tables and o.table_name
not in tables):
+ continue
+ if o.table_name not in contract.table_names:
+ continue
+ t = contract.table(o.table_name)
+ if o.engine is Engine.PYTHON:
+ plan.append(Assignment(t, "prod", o.epoch))
+ if o.shadow_engine is Engine.PYTHON:
+ plan.append(Assignment(t, "shadow", o.epoch))
+ return sorted(plan, key=lambda a: (a.table.table_name, a.target))
+
+
+def in_null_cursor_phase(table: TableContract, start: Watermark | None) ->
bool:
+ return table.cursor_is_usable and (start is None or start.cursor_value is
None)
+
+
+def build_page_query(
+ compat_schema: str,
+ table: TableContract,
+ start: Watermark | None,
+ page_size: int,
+ exclude_pii: bool = False,
+) -> tuple[str, tuple]:
+ pk = quote_ident(table.primary_key)
+ select = []
+ for c in table.columns:
+ if c.synthetic or (exclude_pii and c.pii):
+ select.append(f"CAST(NULL AS {c.type}) AS {quote_ident(c.name)}")
Review Comment:
Fixed.
- Engines: with FBI_EXCLUDE_PII=true the Python engine marks PII columns
unavailable, exactly like Kafka, so both keep the stored value instead of
writing NULL.
- Gate: the shadow reconciliation no longer compares pii columns, so rows
loaded earlier don't block it. meta.scrub_pii() (admin only) clears old values.
- Hash key: the HMAC key is now generated inside the warehouse and only the
control-plane role can read it, so it no longer appears in SQL text, logs or
target/. meta.rotate_pii_hash_key() rotates it.
- Age band: age_band now says Not collected when the date of birth is
excluded, instead of a misleading Unknown.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
