airajena commented on code in PR #102:
URL:
https://github.com/apache/fineract-business-intelligence/pull/102#discussion_r4318481844
##########
extractor/extractor.py:
##########
@@ -37,626 +66,470 @@ 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 build_page_query(
+ compat_schema: str, table: TableContract, start: Watermark | None,
page_size: int
+) -> tuple[str, tuple]:
+ pk = quote_ident(table.primary_key)
+ select = []
+ for c in table.columns:
+ if c.synthetic:
+ select.append(f"CAST(NULL AS {c.type}) AS {quote_ident(c.name)}")
+ elif table.cursor_is_usable and c.name == table.python_cursor:
+ select.append(
+ f"COALESCE({quote_ident(c.name)}, TIMESTAMPTZ '1970-01-01
00:00:00+00') AS {quote_ident(c.name)}"
+ )
+ else:
+ select.append(quote_ident(c.name))
+ base = f"SELECT {', '.join(select)} FROM
{quote_ident(compat_schema)}.{quote_ident(table.table_name)}"
+ if table.cursor_is_usable:
+ cur = f"COALESCE({quote_ident(table.python_cursor)}, TIMESTAMPTZ
'1970-01-01 00:00:00+00')"
Review Comment:
Fixed. The keyset query now uses the raw cursor column, with a separate
first pass for NULL cursors. The docs say an index on (cursor, pk) is needed at
the source. Each (source, table, target) now runs under an advisory lock, and
delete candidates are re-checked against the source right before they're
soft-deleted.
--
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]
