feat: Merged chunk processing with API calls and DB insertion, replac…

…ed Parquet with CSV
MTES-MCT · Feb 24, 2025 · 12c6f5a · 12c6f5a
1 parent dbfa36a
commit 12c6f5a
Show file tree

Hide file tree

Showing 4 changed files with 165 additions and 230 deletions.
diff --git a/.talismanrc b/.talismanrc
@@ -1,10 +1,10 @@
 fileignoreconfig:
-- filename: .github/workflows/review-app.yml
-  checksum: f7719ba0d36160d97e80ee15cb5415b601354576929e36df0596c7d192465cfb
-- filename: README.md
-  checksum: df312ccb4c75fc4c2441a1f7f2c7817ee98ffb3065c78d5d7d6addf6ab129176
-- filename: analytics/.env.example
-  checksum: 917ebeff9022b2a0a1a916ceab18ce47a7fe04ebdbcdeadb62f49167b15b2bde
+- filename: analytics/dagster/src/assets/populate_edited_owners_ban_addresses.py
+  checksum: a8d5d783384c35462f4d2bb022976c9476cf23effc720ba3ecf46d08bae07387
+- filename: analytics/dagster/src/assets/populate_owners_ban_addresses.py
+  checksum: 98cbd0849c6325704a4105e764c79b622c228bbbcd010fb0a554948ab1832fa4
+version: ""
+167b15b2bde
 - filename: analytics/dagster/src/assets/dwh/ingest/ingest_postgres_asset.py
   checksum: b379fbb88e46511c62682abf4cb422e730c4210f6dadc9d5ca648dfd68ff8f88
 - filename: analytics/dagster/src/assets/populate_edited_owners_ban_addresses.py

diff --git a/analytics/dagster/src/assets/populate_edited_owners_ban_addresses.py b/analytics/dagster/src/assets/populate_edited_owners_ban_addresses.py
@@ -5,17 +5,23 @@
 
 @asset(
   description="Return edited owners (score = 1).",
-  required_resource_keys={"psycopg2_connection"}
+  required_resource_keys={"psycopg2_connection", "ban_config"}
 )
 def owners_with_edited_address(context: AssetExecutionContext):
-    query = """
+    config = context.resources.ban_config
+    chunk_size = config.chunk_size
+    max_files = config.max_files
+    disable_max_files = config.disable_max_files
+    query = f"""
     SELECT
         o.id as owner_id,
         array_to_string(o.address_dgfip, ' ') as address_dgfip
     FROM owners o
     LEFT JOIN ban_addresses ba ON o.id = ba.ref_id
-    WHERE ba.ref_id IS NOT NULL AND ba.score = 1;  -- Propriétaires avec adresse éditée par Stéphanie
+    WHERE ba.ref_id IS NOT NULL AND ba.score = 1  -- Propriétaires avec adresse éditée par Stéphanie
+    {"LIMIT " + str(max_files * chunk_size) if not disable_max_files else ""}
     """
+    context.log.info(f"Limit applied: {'LIMIT ' + str(max_files * chunk_size) if not disable_max_files else 'No limit'}")
 
     try:
         with context.resources.psycopg2_connection as conn: