Feature api - #1
Conversation
…ration - Add CSV column validation logic in a new validators.py module. - Enhance database schema to track CSV uploads with a new csv_uploads table. - Create GCS storage functions for uploading and fetching CSV files. - Implement Temporal activities for CSV processing, including fetching, validating, and pushing to Kafka. - Introduce new workflows for orchestrating CSV processing and batch uploads. - Update main application to include new CSV upload routes and health checks. - Modify requirements.txt to include necessary dependencies for GCS and Kafka. - Add tests for Kafka event handling and ensure compatibility with new features.
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughThis pull request adds an authenticated CSV upload pipeline with GCS storage, PostgreSQL tracking, validation, Kafka publishing, and Temporal workflows. It also updates configuration, database lifecycle handling, submission normalization, PII prompts, media persistence, and API startup wiring. ChangesCSV ingestion pipeline
Data quality and initialization updates
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Note
Due to the large number of review comments, Critical severity comments were prioritized as inline comments.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
app/database/db.py (1)
88-99: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winClose and clear the pool when schema initialization fails.
If
create_pool()succeeds butinitialize_schema()fails, the exception path leavesself.poolopen and non-null. A laterconnect()returns immediately without retrying initialization.Proposed fix
except Exception as e: + failed_pool = self.pool + self.pool = None self._ref_count -= 1 + if failed_pool: + try: + await failed_pool.close() + except Exception: + logger.exception("Failed to close partially initialized database pool") logger.error(f"Failed to create database connection pool: {e}") raise🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/database/db.py` around lines 88 - 99, Update the exception handling in the pool initialization flow around create_pool and initialize_schema to close the successfully created pool and clear self.pool before re-raising any failure from schema initialization. Preserve the reference-count decrement and error logging, and ensure later connect calls can retry initialization.
🟠 Major comments (21)
app/database/db.py-36-73 (1)
36-73: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftDo not accept a partially initialized schema.
Using only
submissionsas the initialization marker skipsschema.sqlfor existing deployments, so newly added tables such ascsv_uploadsare never created. Suppressing migration and seed failures then allows startup with missing columns or prompts.Use versioned/idempotent migrations and fail initialization on unexpected migration or seed errors. As per path instructions, review the database layer for transaction safety and rollback handling.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/database/db.py` around lines 36 - 73, The schema initialization logic around the database setup must not use only the submissions table as its completeness marker or suppress migration and seed failures. Replace this with versioned, idempotent migrations that create all current schema objects, run migrations and both seed scripts transactionally, and propagate unexpected errors so the transaction rolls back and initialization fails; preserve safe handling only for explicitly expected concurrent cases.Source: Path instructions
seed_prompts.sql-405-407 (1)
405-407: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winDo not return original PII in persisted analysis metadata.
The prompt requires
pii_found[].textto contain the original identifier, while the activity stores the complete response inmeta_data. This retains the sensitive value that the masking pipeline is intended to remove. Omit the raw text or replace it with non-sensitive type/confidence metadata.Proposed output contract
"pii_found": [ - {"type": "PERSON|LOCATION|ID|PHONE", "text": "original text", "confidence": 0.0, "reason": "max 8 words"} + {"type": "PERSON|LOCATION|ID|PHONE", "confidence": 0.0, "reason": "max 8 words"} ],🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@seed_prompts.sql` around lines 405 - 407, Update the pii_found output contract in the seed prompt so each entry omits the original value from text and retains only non-sensitive metadata such as type, confidence, and reason; ensure persisted meta_data cannot contain raw PII while preserving the analysis result structure.app/database/operations.py-134-135 (1)
134-135: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winValidate that cleaned submission data remains an object.
Kafka only guarantees that the outer event is a dictionary. Values such as
"null"or a list cause_clean_val()to returnNoneor a list, after whichupsert_metadata()anddata.get()fail.Proposed fix
- raw_data = event_payload.get("data", {}) or {} - data = _clean_val(raw_data) + raw_data = event_payload.get("data") + if raw_data is None: + data = {} + elif not isinstance(raw_data, dict): + raise ValueError("Submission data must be a JSON object") + else: + data = _clean_val(raw_data)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/database/operations.py` around lines 134 - 135, Validate the result of _clean_val(raw_data) before passing it to upsert_metadata() or calling data.get(), ensuring submission data is a dictionary/object. For null, list, or other non-object cleaned values, handle the invalid payload through the existing failure path instead of dereferencing it; preserve normal processing for valid dictionaries.app/temporal/pii_and_abusive_activity.py-126-129 (1)
126-129: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winRestrict the fallback to allowed tags and handle multiline spans.
The current pattern does not match wrapped content containing newlines, leaving original PII unmasked. Conversely,
\w+can erase unrelated user-authored XML-like content. Match only documented masking tags and enableDOTALL.Proposed fix
- masked_text = re.sub(r'<\s*(\w+)\s*>.*?<\s*/\s*\1\s*>', r'<\1>', str(masked_text)) + masked_text = re.sub( + r"<\s*(PERSON|PHONE|ID|LOCATION|INSULT|PROFANITY|THREAT)\s*>" + r".*?<\s*/\s*\1\s*>", + lambda match: f"<{match.group(1).upper()}>", + str(masked_text), + flags=re.IGNORECASE | re.DOTALL, + )As per path instructions, review Temporal code for failure recovery.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/pii_and_abusive_activity.py` around lines 126 - 129, Update the fallback regex in the PII masking flow to match multiline wrapped spans by enabling DOTALL, and restrict the tag name to the documented masking tags only (such as INSULT and ID) instead of any \w+ XML-like tag. Preserve the existing replacement behavior that keeps only the allowed tag.Source: Path instructions
app/csv_pipeline/processor.py-116-130 (1)
116-130: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winDo not replace missing source dates with ingestion time.
Empty dates are silently converted to the current time, falsely representing when the submission occurred. Reject the row or preserve
Noneso validation can place the upload on hold.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/csv_pipeline/processor.py` around lines 116 - 130, Update format_datetime so missing values detected by pd.isna or val is None are preserved as None or rejected instead of being replaced with datetime.utcnow(). Ensure downstream validation can place rows with absent source dates on hold, while retaining existing formatting for valid dates.app/temporal/csv_processing_activity.py-156-168 (1)
156-168: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winAvoid random metadata identifiers on fallback.
Every retry or reprocessing attempt generates different program and leader IDs, causing identical CSV rows to reference different entities. Fail processing when mappings are required, or derive deterministic UUIDs from stable tenant/type/name values.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/csv_processing_activity.py` around lines 156 - 168, Update the fallback metadata construction in the CSV processing activity so program and leader identifiers remain stable across retries and reprocessing. Either fail processing when the required mappings are unavailable, or derive deterministic UUIDs using stable tenant, entity type, and name values; do not use uuid.uuid4() for the fallback IDs in the leader_info and program_info blocks.app/temporal/csv_processing_activity.py-214-219 (1)
214-219: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winAtomically claim a bounded batch instead of listing every pending row.
list_by_status("pending")neither appliesCRON_BATCH_SIZEnor changes status. Overlapping schedules can therefore start duplicate workflows for the same records. Use the existingclaim_pending_records(settings.CRON_BATCH_SIZE)operation.Proposed fix
async def fetch_pending_csv_uploads_activity() -> List[int]: - records = await csv_upload_repo.list_by_status("pending") + records = await csv_upload_repo.claim_pending_records( + settings.CRON_BATCH_SIZE + ) return [r["id"] for r in records]As per path instructions, review Temporal code for proper workflow orchestration and failure recovery.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/csv_processing_activity.py` around lines 214 - 219, Update fetch_pending_csv_uploads_activity to call the existing csv_upload_repo.claim_pending_records operation with settings.CRON_BATCH_SIZE instead of list_by_status("pending"), and return the claimed record IDs. Preserve the activity’s async behavior and ensure claiming atomically transitions records so overlapping schedules cannot select the same uploads.Source: Path instructions
app/csv_pipeline/processor.py-147-153 (1)
147-153: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winReject rows without a valid submission ID.
A missing ID becomes
None; later fallbacks produceuserId="None". Multiple malformed rows then share invalid identifiers in Kafka. Validate the ID and fail the row before constructing the payload.Proposed fix
- try: - submission_id = int(row_dict.get("id")) - except Exception: - submission_id = row_dict.get("id") + raw_submission_id = row_dict.get("id") + if raw_submission_id is None: + raise ValueError("CSV row is missing required field 'id'") + try: + submission_id = int(raw_submission_id) + except (TypeError, ValueError) as exc: + raise ValueError(f"Invalid submission id: {raw_submission_id!r}") from exc🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/csv_pipeline/processor.py` around lines 147 - 153, Update the row-processing logic around submission_id to validate that row_dict.get("id") is present and convertible to a valid identifier before constructing the payload. Reject or skip the row when the ID is missing or malformed, rather than falling back to None or the original invalid value; preserve normal processing for valid IDs.app/csv_pipeline/csv_upload_repo.py-51-80 (1)
51-80: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winPersist the status selected by the upload endpoint.
app/api/csv_upload.pyselectson_holdfor invalid CSVs, but this function always insertspending. Consequently, batch processing can claim files the API reported as on hold.Proposed fix
async def insert_upload_record( report_type: str, program_name: str, leader_category: str, cloud_storage_path: str, file_name: str | None = None, file_size: int | None = None, meta_data: dict[str, Any] | None = None, + status: str = "pending", ) -> int: - """Insert a new row with status='pending'. Returns the new row's id.""" + """Insert a new upload record and return its id.""" @@ - VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, 'pending') + VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8) @@ json.dumps(meta_data or {}), + status,Pass the endpoint's computed
statuswhen calling this function.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/csv_pipeline/csv_upload_repo.py` around lines 51 - 80, Update insert_upload_record to accept the upload status as an argument and use it in the INSERT instead of the hardcoded 'pending' value. Update the app/api/csv_upload.py call site to pass its computed status, preserving on_hold for invalid CSVs and pending for valid uploads.app/temporal/csv_processing_activity.py-111-113 (1)
111-113: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy liftMove the blocking CSV/GCS/Kafka work off these async activities.
fetch_csv,load_csv,future.get, and_flush_producer()are synchronous, so they can stall the Temporal worker loop while uploads are processed. Use a sync activity or wrap the blocking sections withasyncio.to_thread. This affectscsv_fetch_and_validate_activityandcsv_push_to_kafka_activity.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/csv_processing_activity.py` around lines 111 - 113, Update csv_fetch_and_validate_activity and csv_push_to_kafka_activity so blocking operations do not run directly on the async Temporal worker loop: move fetch_csv, load_csv, future.get, and _flush_producer into synchronous activities or execute them via asyncio.to_thread, while preserving the existing CSV validation and Kafka upload behavior.Source: Path instructions
app/temporal/csv_processing_activity.py-23-44 (1)
23-44: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftMake row publishing idempotent. Temporal retries are already disabled for this activity, but the Kafka producer can still duplicate an ambiguously acknowledged send, and any rerun after an
on_holdfailure will republish earlier rows. Add a persisted per-row checkpoint/outbox or downstream dedupe keyed by record + row.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/csv_processing_activity.py` around lines 23 - 44, Make _push_row idempotent by adding durable per-record-and-row checkpoint/outbox state, or enforce equivalent downstream deduplication using a stable record-plus-row key. Check this state before sending, atomically record successful publication, and ensure reruns after on_hold skip rows already published while safely handling ambiguous Kafka acknowledgements.Source: Path instructions
app/temporal/csv_processing_activity.py-65-78 (1)
65-78: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winRe-raise GCS fetch failures; keep
Falsefor deterministic CSV issues. ReturningFalsehere makes transient storage/network errors finish the activity normally, so the Temporal retry policy inCsvProcessingWorkflownever runs. Splitfetch_csv()/load_csv()errors from parse/validation failures and raise on the former so retries can recover.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/csv_processing_activity.py` around lines 65 - 78, Update the CSV processing activity around fetch_csv() and load_csv() to distinguish transient GCS retrieval failures from deterministic CSV parsing or validation failures. Propagate or re-raise exceptions originating from fetch_csv() so CsvProcessingWorkflow’s Temporal retry policy can run, while retaining the existing on_hold status update and False return for load_csv() parse/validation failures.Source: Path instructions
app/api/csv_upload.py-101-116 (1)
101-116: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winPersist the computed validation status.
statusbecomeson_holdfor an invalid CSV, butinsert_upload_record()always insertspending. The response and database therefore disagree, and batch processing can consume invalid uploads. Passstatusinto the repository insert.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/api/csv_upload.py` around lines 101 - 116, Pass the computed status from the validation branch into insert_upload_record() in the CSV upload flow, so invalid files persist as on_hold and valid files as pending. Update the repository method and its underlying insert mapping to accept and store this status while preserving existing metadata and response behavior.schema.sql-379-392 (1)
379-392: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winEnforce duplicate-file rejection atomically in PostgreSQL.
The application-level check can race because the database does not enforce the same invariant.
schema.sql#L379-L392: add a unique constraint/index coveringprogram_name,leader_category,report_type,file_name, andfile_size.app/api/csv_upload.py#L63-L72: retain the check only as an optimization, catch the uniqueness violation, and return HTTP 409.As per path instructions, SQL must use proper constraints and FastAPI endpoints must return correct HTTP status codes.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@schema.sql` around lines 379 - 392, Atomically enforce duplicate CSV rejection by adding a proper unique constraint covering program_name, leader_category, report_type, file_name, and file_size in schema.sql. In app/api/csv_upload.py, retain the existing duplicate check only as an optimization, catch the database uniqueness violation from the insert/create operation, and return an HTTP 409 response for concurrent duplicates.Source: Path instructions
app/csv_pipeline/validators.py-31-40 (1)
31-40: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winValidate every configured column before calling
.lower().A valid JSON array containing
null, numbers, or objects passes the current check and then raisesAttributeError, instead of returning validation errors.Proposed fix
expected_cols = json.loads(raw_cols) - if not isinstance(expected_cols, list): - raise ValueError("Expected columns must be a JSON array") + if not isinstance(expected_cols, list) or not all( + isinstance(column, str) and column + for column in expected_cols + ): + raise ValueError("Expected columns must be a JSON array of non-empty strings")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/csv_pipeline/validators.py` around lines 31 - 40, Update the expected_cols validation in the JSON parsing block to verify every configured column is a string before expected_cols_lower calls .lower(). Return the existing validation-error tuple with a clear message for any invalid element, while preserving the current JSON-array and case-insensitive comparison behavior for valid input.app/api/csv_upload.py-123-146 (1)
123-146: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winSurface real-time workflow start failures instead of returning success. The
start_workflow()exception is only logged here, so a real-time upload can return200while the record stayspendingindefinitely. Mark the record as failed/on_hold and return a 5xx, or move the workflow start onto a durable enqueue path.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/api/csv_upload.py` around lines 123 - 146, Update the real-time workflow-start error path in the upload handler around Client.connect and start_workflow so failures are surfaced instead of returning the normal success response. On exception, mark the associated record as failed or on_hold using the existing persistence flow, then return an appropriate 5xx response; alternatively, route the start request through an existing durable enqueue mechanism. Preserve the current success response only when the workflow starts successfully.Source: Path instructions
app/api/csv_upload.py-56-60 (1)
56-60: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy liftBound the upload before reading it into memory.
await file.read()loads the entire CSV at once, andpd.read_csv(io.BytesIO(file_bytes))creates another in-memory copy. Enforce a maximum upload size and return HTTP 413 when it’s exceeded.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/api/csv_upload.py` around lines 56 - 60, Update the upload handling in the CSV endpoint around file.read and file_size to enforce a maximum size before loading the entire file into memory, returning HTTP 413 when the limit is exceeded. Use the existing upload configuration or define a single bounded-read mechanism, and preserve the current empty-file HTTP 400 behavior for accepted uploads.Source: Path instructions
app/api/csv_upload.py-158-175 (1)
158-175: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winRestrict manual retries to explicit source states.
This endpoint currently moves any non-in_progressrecord — includingprocessed— back toin_progressbefore Temporal accepts the run. Limit retries to the intended states (for example,pending/on_hold) and only persist the new status once the workflow has started successfully, or roll back to the prior state on failure.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/api/csv_upload.py` around lines 158 - 175, Restrict the retry logic around the CSV upload endpoint to records in the allowed source states, such as pending or on_hold, and reject processed or other states before changing status. Start the Temporal workflow first, then persist in_progress only after successful acceptance; on failure, preserve or restore the record’s prior status rather than unconditionally setting on_hold. Update the status handling in the workflow-start block while retaining the existing CsvProcessingWorkflow invocation.app/storage/gcs.py-65-68 (1)
65-68: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winAvoid blocking async handlers with sync GCS calls
upload_from_string()anddownload_as_bytes()block the FastAPI route and the async Temporal activity worker here; offload them to a thread or move the GCS I/O into sync activities, and pass an explicit timeout if you need a shorter bound.app/storage/gcs.py:68,85🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/storage/gcs.py` around lines 65 - 68, Update the GCS upload and download operations in the relevant storage functions to avoid blocking async handlers: offload the synchronous blob.upload_from_string() and blob.download_as_bytes() calls to a worker thread, or move them into synchronous activities. Preserve existing behavior and provide an explicit timeout when a shorter operation bound is required.app/config.py-54-60 (1)
54-60: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winValidate the CSV schema settings before they are used.
STORY_CSV_COLUMNandDISCUSSION_CSV_COLUMNare still raw JSON strings, andvalidate_columns()assumes each entry is a unique, non-empty string. A bad env value can fail the Temporal validation path instead of surfacing as a config error; a small settings validator/shared parser would make this fail fast.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/config.py` around lines 54 - 60, Add shared settings validation for STORY_CSV_COLUMN and DISCUSSION_CSV_COLUMN that parses each JSON string and verifies it is a non-empty list of unique, non-empty strings before configuration is accepted. Reuse this parser or validator from validate_columns() so malformed environment values fail during settings initialization as configuration errors rather than during Temporal validation.Source: Path instructions
main.py-123-127 (1)
123-127: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winPrevent application hang during graceful shutdown.
When
run_web_async()(which wraps Uvicorn) runs concurrently with infinite tasks likerun_consumerandrun_workerviaasyncio.gather, a shutdown signal (Ctrl+C) will be intercepted by Uvicorn. Uvicorn will shut down gracefully and exit its task, but it suppresses theKeyboardInterrupt. Consequently,gatherwill hang indefinitely waiting for the consumer and worker tasks to finish.To fix this, use
asyncio.waitwithFIRST_COMPLETEDso that when Uvicorn shuts down, the sibling tasks are explicitly cancelled.Note: Since
run_consumer()currently relies onKeyboardInterruptfor its shutdown logic, it will not cleanly executeconsumer.stop()when cancelled viaasyncio.CancelledError. Consider updatingrun_consumerto handleasyncio.CancelledErroror use afinallyblock to ensure the Kafka consumer cleanly leaves the group.🔄 Proposed fix for graceful task termination
async def run_all_services(): global consumer_running, worker_running consumer_running = True worker_running = True - await asyncio.gather(run_web_async(), run_consumer(), run_worker()) + + web_task = asyncio.create_task(run_web_async()) + consumer_task = asyncio.create_task(run_consumer()) + worker_task = asyncio.create_task(run_worker()) + + done, pending = await asyncio.wait( + [web_task, consumer_task, worker_task], + return_when=asyncio.FIRST_COMPLETED + ) + + # Cancel the remaining tasks so the application can exit cleanly + for task in pending: + task.cancel() + + # Await the cancelled tasks to ensure they clean up + if pending: + await asyncio.gather(*pending, return_exceptions=True)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@main.py` around lines 123 - 127, Update run_all_services to create tasks for run_web_async, run_consumer, and run_worker, then use asyncio.wait with return_when=asyncio.FIRST_COMPLETED; cancel and await all unfinished sibling tasks when any service exits. Update run_consumer so cancellation via asyncio.CancelledError still executes consumer.stop(), preferably through shared cleanup or a finally block, while preserving its existing shutdown behavior.
🟡 Minor comments (1)
app/services/gcp_storage.py-52-60 (1)
52-60: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winNormalize empty prefixes and leading blob separators.
With
BUCKET_NAME="bucket/",prefixbecomes/, creating an object named/blob. Normalize both components before constructing the object path.Proposed fix
bucket_name = settings.BUCKET_NAME prefix = "" if "/" in bucket_name: - parts = bucket_name.split("/", 1) - bucket_name = parts[0] - prefix = parts[1].strip("/") + "/" + bucket_name, configured_prefix = bucket_name.split("/", 1) + configured_prefix = configured_prefix.strip("/") + if configured_prefix: + prefix = f"{configured_prefix}/" bucket = client.bucket(bucket_name) - full_blob_name = f"{prefix}{blob_name}" + full_blob_name = f"{prefix}{blob_name.lstrip('/')}"🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/services/gcp_storage.py` around lines 52 - 60, Update the bucket parsing and object-name construction in the GCP storage method to normalize empty prefixes and leading separators: ensure a trailing slash in settings.BUCKET_NAME produces an empty prefix, and remove leading “/” characters from blob_name before combining it with prefix. Preserve the existing bucket extraction and construct full_blob_name without an unintended leading slash.
🧹 Nitpick comments (3)
app/temporal/worker.py (1)
82-123: 📐 Maintainability & Code Quality | 🔵 TrivialSchedule updates via configuration changes.
The worker catches
ScheduleAlreadyRunningErrorand skips registration if the schedule exists. If the cron expressions (settings.CSV_SCHEDULE_CRON_TIMEorsettings.BATCH_SCHEDULE_CRON) are later modified in the environment, the new schedules will not take effect automatically because the existing schedule blocks it.Consider handling schedule updates (e.g., fetching the handle and updating it) if dynamic cron configuration is required, or explicitly document that schedules must be manually deleted via the Temporal CLI/UI to apply configuration changes.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/worker.py` around lines 82 - 123, Update the schedule registration logic around the CSV and analysis schedule creation blocks to handle existing schedules when their configured cron expressions change: fetch each existing schedule handle and update its specification to the current settings before skipping registration. If dynamic updates are not supported, explicitly document that users must delete the existing Temporal schedules through the CLI or UI for configuration changes to take effect.app/temporal/workflows.py (2)
253-258: 🗄️ Data Integrity & Integration | 🔵 TrivialIdempotency risk on manual retry.
While
maximum_attempts=1prevents automatic Temporal retries from duplicating Kafka writes, a manual workflow retry (e.g., via the/push/{record_id}endpoint) will start a new workflow execution that processes and pushes the CSV from the beginning.As per path instructions, ensure activities are idempotent. Verify that the downstream Kafka consumers can safely handle duplicate messages (e.g., using the
f"{record_id}-{pushed}"Kafka key for upserts), or consider checkpointing therows_pushedcount to resume safely during manual retries.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/workflows.py` around lines 253 - 258, Ensure the workflow using csv_push_to_kafka_activity is safe across manual retries, not only automatic retries: verify or enforce deterministic Kafka keys such as f"{record_id}-{pushed}" so consumers upsert duplicate messages, or persist and reuse the rows_pushed checkpoint to resume without replaying already-pushed rows. Preserve the existing single-attempt RetryPolicy while updating the activity or downstream handling as needed.Source: Path instructions
301-307: 🩺 Stability & Availability | 🔵 TrivialHandle overlapping child workflows on retry.
When fanning out child workflows, if the
CsvBatchProcessingWorkflowitself fails and retries, it may attempt to start child workflows that are already running from the previous attempt (if the DB still shows them aspending). Temporal will reject the duplicate child workflow ID, causing the batch to fail again.As per path instructions, ensure proper workflow orchestration and failure recovery. Consider catching
WorkflowExecutionAlreadyStartedErroror using an appropriate ID conflict policy (likeWorkflowIdConflictPolicy.USE_EXISTING) so that retries simply attach to the already-running child workflows.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/workflows.py` around lines 301 - 307, Update the child-workflow fan-out in CsvBatchProcessingWorkflow.run to handle retries when a child with the same ID is already running. Configure the child start with the appropriate Temporal ID conflict policy, such as USE_EXISTING, or catch WorkflowExecutionAlreadyStartedError and reuse the existing execution, while preserving normal execution for new child workflows.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@app/api/csv_upload.py`:
- Around line 149-150: Update the push_record endpoint to require the existing
authentication and authorization dependency, matching the protection used by the
/upload/ endpoint. Ensure unauthenticated or unauthorized callers cannot
initiate processing or mutate upload status.
In `@app/config.py`:
- Around line 49-50: Update the AUTH_TOKEN configuration and the nearby
BUCKET_NAME setting to require values from the environment, removing their
hardcoded/default fallbacks. Ensure missing auth or bucket configuration causes
startup validation to fail rather than using repository-visible defaults.
---
Outside diff comments:
In `@app/database/db.py`:
- Around line 88-99: Update the exception handling in the pool initialization
flow around create_pool and initialize_schema to close the successfully created
pool and clear self.pool before re-raising any failure from schema
initialization. Preserve the reference-count decrement and error logging, and
ensure later connect calls can retry initialization.
---
Major comments:
In `@app/api/csv_upload.py`:
- Around line 101-116: Pass the computed status from the validation branch into
insert_upload_record() in the CSV upload flow, so invalid files persist as
on_hold and valid files as pending. Update the repository method and its
underlying insert mapping to accept and store this status while preserving
existing metadata and response behavior.
- Around line 123-146: Update the real-time workflow-start error path in the
upload handler around Client.connect and start_workflow so failures are surfaced
instead of returning the normal success response. On exception, mark the
associated record as failed or on_hold using the existing persistence flow, then
return an appropriate 5xx response; alternatively, route the start request
through an existing durable enqueue mechanism. Preserve the current success
response only when the workflow starts successfully.
- Around line 56-60: Update the upload handling in the CSV endpoint around
file.read and file_size to enforce a maximum size before loading the entire file
into memory, returning HTTP 413 when the limit is exceeded. Use the existing
upload configuration or define a single bounded-read mechanism, and preserve the
current empty-file HTTP 400 behavior for accepted uploads.
- Around line 158-175: Restrict the retry logic around the CSV upload endpoint
to records in the allowed source states, such as pending or on_hold, and reject
processed or other states before changing status. Start the Temporal workflow
first, then persist in_progress only after successful acceptance; on failure,
preserve or restore the record’s prior status rather than unconditionally
setting on_hold. Update the status handling in the workflow-start block while
retaining the existing CsvProcessingWorkflow invocation.
In `@app/config.py`:
- Around line 54-60: Add shared settings validation for STORY_CSV_COLUMN and
DISCUSSION_CSV_COLUMN that parses each JSON string and verifies it is a
non-empty list of unique, non-empty strings before configuration is accepted.
Reuse this parser or validator from validate_columns() so malformed environment
values fail during settings initialization as configuration errors rather than
during Temporal validation.
In `@app/csv_pipeline/csv_upload_repo.py`:
- Around line 51-80: Update insert_upload_record to accept the upload status as
an argument and use it in the INSERT instead of the hardcoded 'pending' value.
Update the app/api/csv_upload.py call site to pass its computed status,
preserving on_hold for invalid CSVs and pending for valid uploads.
In `@app/csv_pipeline/processor.py`:
- Around line 116-130: Update format_datetime so missing values detected by
pd.isna or val is None are preserved as None or rejected instead of being
replaced with datetime.utcnow(). Ensure downstream validation can place rows
with absent source dates on hold, while retaining existing formatting for valid
dates.
- Around line 147-153: Update the row-processing logic around submission_id to
validate that row_dict.get("id") is present and convertible to a valid
identifier before constructing the payload. Reject or skip the row when the ID
is missing or malformed, rather than falling back to None or the original
invalid value; preserve normal processing for valid IDs.
In `@app/csv_pipeline/validators.py`:
- Around line 31-40: Update the expected_cols validation in the JSON parsing
block to verify every configured column is a string before expected_cols_lower
calls .lower(). Return the existing validation-error tuple with a clear message
for any invalid element, while preserving the current JSON-array and
case-insensitive comparison behavior for valid input.
In `@app/database/db.py`:
- Around line 36-73: The schema initialization logic around the database setup
must not use only the submissions table as its completeness marker or suppress
migration and seed failures. Replace this with versioned, idempotent migrations
that create all current schema objects, run migrations and both seed scripts
transactionally, and propagate unexpected errors so the transaction rolls back
and initialization fails; preserve safe handling only for explicitly expected
concurrent cases.
In `@app/database/operations.py`:
- Around line 134-135: Validate the result of _clean_val(raw_data) before
passing it to upsert_metadata() or calling data.get(), ensuring submission data
is a dictionary/object. For null, list, or other non-object cleaned values,
handle the invalid payload through the existing failure path instead of
dereferencing it; preserve normal processing for valid dictionaries.
In `@app/storage/gcs.py`:
- Around line 65-68: Update the GCS upload and download operations in the
relevant storage functions to avoid blocking async handlers: offload the
synchronous blob.upload_from_string() and blob.download_as_bytes() calls to a
worker thread, or move them into synchronous activities. Preserve existing
behavior and provide an explicit timeout when a shorter operation bound is
required.
In `@app/temporal/csv_processing_activity.py`:
- Around line 156-168: Update the fallback metadata construction in the CSV
processing activity so program and leader identifiers remain stable across
retries and reprocessing. Either fail processing when the required mappings are
unavailable, or derive deterministic UUIDs using stable tenant, entity type, and
name values; do not use uuid.uuid4() for the fallback IDs in the leader_info and
program_info blocks.
- Around line 214-219: Update fetch_pending_csv_uploads_activity to call the
existing csv_upload_repo.claim_pending_records operation with
settings.CRON_BATCH_SIZE instead of list_by_status("pending"), and return the
claimed record IDs. Preserve the activity’s async behavior and ensure claiming
atomically transitions records so overlapping schedules cannot select the same
uploads.
- Around line 111-113: Update csv_fetch_and_validate_activity and
csv_push_to_kafka_activity so blocking operations do not run directly on the
async Temporal worker loop: move fetch_csv, load_csv, future.get, and
_flush_producer into synchronous activities or execute them via
asyncio.to_thread, while preserving the existing CSV validation and Kafka upload
behavior.
- Around line 23-44: Make _push_row idempotent by adding durable
per-record-and-row checkpoint/outbox state, or enforce equivalent downstream
deduplication using a stable record-plus-row key. Check this state before
sending, atomically record successful publication, and ensure reruns after
on_hold skip rows already published while safely handling ambiguous Kafka
acknowledgements.
- Around line 65-78: Update the CSV processing activity around fetch_csv() and
load_csv() to distinguish transient GCS retrieval failures from deterministic
CSV parsing or validation failures. Propagate or re-raise exceptions originating
from fetch_csv() so CsvProcessingWorkflow’s Temporal retry policy can run, while
retaining the existing on_hold status update and False return for load_csv()
parse/validation failures.
In `@app/temporal/pii_and_abusive_activity.py`:
- Around line 126-129: Update the fallback regex in the PII masking flow to
match multiline wrapped spans by enabling DOTALL, and restrict the tag name to
the documented masking tags only (such as INSULT and ID) instead of any \w+
XML-like tag. Preserve the existing replacement behavior that keeps only the
allowed tag.
In `@main.py`:
- Around line 123-127: Update run_all_services to create tasks for
run_web_async, run_consumer, and run_worker, then use asyncio.wait with
return_when=asyncio.FIRST_COMPLETED; cancel and await all unfinished sibling
tasks when any service exits. Update run_consumer so cancellation via
asyncio.CancelledError still executes consumer.stop(), preferably through shared
cleanup or a finally block, while preserving its existing shutdown behavior.
In `@schema.sql`:
- Around line 379-392: Atomically enforce duplicate CSV rejection by adding a
proper unique constraint covering program_name, leader_category, report_type,
file_name, and file_size in schema.sql. In app/api/csv_upload.py, retain the
existing duplicate check only as an optimization, catch the database uniqueness
violation from the insert/create operation, and return an HTTP 409 response for
concurrent duplicates.
In `@seed_prompts.sql`:
- Around line 405-407: Update the pii_found output contract in the seed prompt
so each entry omits the original value from text and retains only non-sensitive
metadata such as type, confidence, and reason; ensure persisted meta_data cannot
contain raw PII while preserving the analysis result structure.
---
Minor comments:
In `@app/services/gcp_storage.py`:
- Around line 52-60: Update the bucket parsing and object-name construction in
the GCP storage method to normalize empty prefixes and leading separators:
ensure a trailing slash in settings.BUCKET_NAME produces an empty prefix, and
remove leading “/” characters from blob_name before combining it with prefix.
Preserve the existing bucket extraction and construct full_blob_name without an
unintended leading slash.
---
Nitpick comments:
In `@app/temporal/worker.py`:
- Around line 82-123: Update the schedule registration logic around the CSV and
analysis schedule creation blocks to handle existing schedules when their
configured cron expressions change: fetch each existing schedule handle and
update its specification to the current settings before skipping registration.
If dynamic updates are not supported, explicitly document that users must delete
the existing Temporal schedules through the CLI or UI for configuration changes
to take effect.
In `@app/temporal/workflows.py`:
- Around line 253-258: Ensure the workflow using csv_push_to_kafka_activity is
safe across manual retries, not only automatic retries: verify or enforce
deterministic Kafka keys such as f"{record_id}-{pushed}" so consumers upsert
duplicate messages, or persist and reuse the rows_pushed checkpoint to resume
without replaying already-pushed rows. Preserve the existing single-attempt
RetryPolicy while updating the activity or downstream handling as needed.
- Around line 301-307: Update the child-workflow fan-out in
CsvBatchProcessingWorkflow.run to handle retries when a child with the same ID
is already running. Configure the child start with the appropriate Temporal ID
conflict policy, such as USE_EXISTING, or catch
WorkflowExecutionAlreadyStartedError and reuse the existing execution, while
preserving normal execution for new child workflows.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: b74b5a91-06bb-4c07-a593-52303334e3be
📒 Files selected for processing (22)
.env.exampleapp/api/csv_upload.pyapp/config.pyapp/csv_pipeline/README.mdapp/csv_pipeline/__init__.pyapp/csv_pipeline/csv_upload_repo.pyapp/csv_pipeline/processor.pyapp/csv_pipeline/validators.pyapp/database/db.pyapp/database/operations.pyapp/services/gcp_storage.pyapp/storage/__init__.pyapp/storage/gcs.pyapp/temporal/csv_processing_activity.pyapp/temporal/pii_and_abusive_activity.pyapp/temporal/worker.pyapp/temporal/workflows.pymain.pyrequirements.txtschema.sqlseed_prompts.sqltests/test_kafka_events.py
…ance manual trigger processing
…nt variable descriptions
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (1)
app/api/routes.py (1)
118-189: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftMove upload orchestration into a service layer.
Duplicate detection, parsing, storage, persistence, and workflow startup are all embedded in the route. Extract them into an ingestion service so the endpoint only validates input and maps service results to HTTP responses.
As per path instructions, FastAPI endpoints require “No business logic inside route handlers.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/api/routes.py` around lines 118 - 189, Refactor the route handler containing the duplicate check, CSV parsing, GCS upload, record insertion, and real-time CsvProcessingWorkflow startup into a dedicated ingestion service method. Keep the endpoint limited to request/input validation, invoking the service, and mapping returned results or service exceptions to HTTP responses; preserve the existing validation status, metadata, persistence, logging, and workflow behavior through the service layer.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@app/api/routes.py`:
- Around line 210-229: Update the record-claim flow before starting
CsvProcessingWorkflow so csv_upload_repo atomically changes the record to
"in_progress" only when its current status is eligible, using a conditional
update that returns the claimed row. In the route, raise HTTP 409 when no row is
returned, and only then connect to Temporal and start the workflow; preserve the
existing failure handling for errors after the claim.
- Around line 129-146: Move the blocking pd.read_csv/validate_columns work and
upload_csv call in the async route to a thread-based async boundary, such as
FastAPI’s run_in_threadpool or an equivalent executor. Preserve the existing
validation error handling and HTTPException behavior while ensuring neither
synchronous operation executes directly on the event loop.
- Around line 177-189: The real-time workflow failure path in the upload handler
must not report a successful upload. In the exception block around
Client.connect and start_workflow, update the record status to on_hold with the
error details, persist that change using the existing record-update mechanism,
and return the route’s appropriate failure response instead of continuing as
successful.
- Around line 155-169: The calculated status in the route is not persisted
because insert_upload_record defaults to pending. Update insert_upload_record
and its call site to accept and store the local status value, ensuring invalid
CSVs remain on_hold and the record insertion persists status atomically.
- Around line 43-44: Update the trigger_submission_manually endpoint to require
authentication using the existing verify_auth_token dependency, while preserving
its current request handling and tenant-scoped workflow behavior.
- Around line 111-115: Update the upload handling around the endpoint’s
file.read call to validate the request body size before reading the entire file
into memory, enforcing the configured maximum upload size through FastAPI
request validation. Preserve the existing empty-file rejection and ensure
oversized uploads are rejected before parsing or allocating their full contents.
In `@app/database/operations.py`:
- Around line 222-226: Update the event payload handling around _clean_val so
the "update" branch assigns event_payload.get("newValues", {}) to raw_data,
matching the existing non-update branch. Preserve the subsequent
_clean_val(raw_data) flow for all event types.
---
Nitpick comments:
In `@app/api/routes.py`:
- Around line 118-189: Refactor the route handler containing the duplicate
check, CSV parsing, GCS upload, record insertion, and real-time
CsvProcessingWorkflow startup into a dedicated ingestion service method. Keep
the endpoint limited to request/input validation, invoking the service, and
mapping returned results or service exceptions to HTTP responses; preserve the
existing validation status, metadata, persistence, logging, and workflow
behavior through the service layer.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 12fdce1a-21d2-44d6-843a-4a5a8b95ab58
📒 Files selected for processing (11)
.env.exampleapp/api/routes.pyapp/config.pyapp/database/operations.pyapp/temporal/pii_and_abusive_activity.pyapp/temporal/worker.pyapp/temporal/workflows.pymain.pyrequirements.txtschema.sqltests/test_kafka_events.py
🚧 Files skipped from review as they are similar to previous changes (7)
- app/temporal/pii_and_abusive_activity.py
- schema.sql
- app/temporal/worker.py
- tests/test_kafka_events.py
- main.py
- app/config.py
- app/temporal/workflows.py
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
app/api/routes.py (3)
183-183: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winReuse Temporal client connection to prevent resource leaks.
Connecting to Temporal on every request (
await Client.connect(...)) without closing the client leaks gRPC connections and degrades performance. The Temporal client is intended to be a long-lived, thread-safe object.
app/api/routes.py#L183-L183: remove per-request initialization and reuse a globally instantiated client (e.g., from FastAPI lifespan events).app/api/routes.py#L226-L226: reuse the same global client instance here instead of creating a new one.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/api/routes.py` at line 183, Replace per-request Temporal client creation in app/api/routes.py lines 183-183 and 226-226 with a shared, long-lived client initialized through the application lifecycle (such as FastAPI lifespan) and reused by both request paths. Ensure the lifecycle manages the client connection and cleanup.
89-95: 📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy liftExtract business logic into a dedicated service layer.
As per path instructions, there should be "No business logic inside route handlers". The current route handlers heavily orchestrate duplicate checking, data validation, cloud storage uploads, database insertions, and Temporal triggers.
app/api/routes.py#L89-L95: extract the file processing, database insertion, and Temporal workflow orchestration into a separate service function.app/api/routes.py#L212-L215: extract the record claiming and Temporal workflow triggering into a separate service function.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/api/routes.py` around lines 89 - 95, The route handlers in app/api/routes.py currently contain business orchestration that must move into dedicated service functions. At app/api/routes.py lines 89-95, extract file processing, validation/duplicate handling, database insertion, and Temporal workflow orchestration from upload_report into a service, leaving the handler responsible only for request concerns and delegating to it. At app/api/routes.py lines 212-215, likewise extract record claiming and Temporal workflow triggering into a separate service function invoked by the corresponding route handler.Source: Path instructions
199-208: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winUse Pydantic models for endpoint responses.
As per path instructions, FastAPI endpoints require "Proper response models". These endpoints currently return raw dictionaries, which bypasses FastAPI's schema validation and OpenAPI documentation generation.
app/api/routes.py#L199-L208: define a Pydantic model (e.g.,UploadResponse) and use it in the route'sresponse_modelparameter instead of returning a raw dictionary.app/api/routes.py#L233-L233: define a Pydantic model for this success response and add it to the route'sresponse_modelparameter.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/api/routes.py` around lines 199 - 208, Define Pydantic response models for both endpoint responses and reference them through each route’s response_model parameter: update the upload response around lines 199-208 in app/api/routes.py to use an UploadResponse-style model that includes the conditional errors field, and update the success response at line 233 in app/api/routes.py with its own appropriate model. Return model instances rather than raw dictionaries so FastAPI performs validation and generates the response schemas.Source: Path instructions
🧹 Nitpick comments (1)
app/api/routes.py (1)
145-149: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winPreserve original exception traces using
from.As per path instructions, endpoints must have "Consistent exception handling". When raising an
HTTPExceptionfrom within anexceptblock, use exception chaining to preserve the original stack trace. This prevents blind exception catching and aids in debugging 500 errors.
app/api/routes.py#L145-L149: appendfrom excto theraisestatement to chain the original exception.app/api/routes.py#L191-L197: appendfrom eto theraisestatement to chain the original exception.app/api/routes.py#L234-L238: appendfrom eto theraisestatement to chain the original exception.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/api/routes.py` around lines 145 - 149, Preserve original exception chaining when converting upload failures to HTTP errors: update the raise statements in app/api/routes.py at lines 145-149, 191-197, and 234-238 to chain each HTTPException from its caught exception variable (`exc` or `e`) using `from`.Sources: Path instructions, Linters/SAST tools
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@app/api/routes.py`:
- Line 183: Replace per-request Temporal client creation in app/api/routes.py
lines 183-183 and 226-226 with a shared, long-lived client initialized through
the application lifecycle (such as FastAPI lifespan) and reused by both request
paths. Ensure the lifecycle manages the client connection and cleanup.
- Around line 89-95: The route handlers in app/api/routes.py currently contain
business orchestration that must move into dedicated service functions. At
app/api/routes.py lines 89-95, extract file processing, validation/duplicate
handling, database insertion, and Temporal workflow orchestration from
upload_report into a service, leaving the handler responsible only for request
concerns and delegating to it. At app/api/routes.py lines 212-215, likewise
extract record claiming and Temporal workflow triggering into a separate service
function invoked by the corresponding route handler.
- Around line 199-208: Define Pydantic response models for both endpoint
responses and reference them through each route’s response_model parameter:
update the upload response around lines 199-208 in app/api/routes.py to use an
UploadResponse-style model that includes the conditional errors field, and
update the success response at line 233 in app/api/routes.py with its own
appropriate model. Return model instances rather than raw dictionaries so
FastAPI performs validation and generates the response schemas.
---
Nitpick comments:
In `@app/api/routes.py`:
- Around line 145-149: Preserve original exception chaining when converting
upload failures to HTTP errors: update the raise statements in app/api/routes.py
at lines 145-149, 191-197, and 234-238 to chain each HTTPException from its
caught exception variable (`exc` or `e`) using `from`.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 8ae3bf0d-c93e-499e-b8fc-a4ae09ae06f2
📒 Files selected for processing (4)
app/api/routes.pyapp/config.pyapp/csv_pipeline/csv_upload_repo.pyapp/database/operations.py
🚧 Files skipped from review as they are similar to previous changes (3)
- app/csv_pipeline/csv_upload_repo.py
- app/database/operations.py
- app/config.py
Vivek-M-08
left a comment
Vivek-M-08
left a comment
There was a problem hiding this comment.
restructure the files as per the attached plan and resolve the comments
| # --------------------------------------------------------------------------- | ||
| # Shared auth | ||
| # --------------------------------------------------------------------------- |
There was a problem hiding this comment.
just keep the comments in one line
| # --------------------------------------------------------------------------- | ||
| # Submissions router → /api/submissions | ||
| # --------------------------------------------------------------------------- |
| if not file.filename or not file.filename.lower().endswith(".csv"): | ||
| raise HTTPException(status_code=400, detail="Only .csv files are accepted") | ||
|
|
||
| file_bytes = await file.read(settings.MAX_CSV_UPLOAD_BYTES + 1) |
There was a problem hiding this comment.
MAX_CSV_UPLOAD_BYTES is missing in .env.example
| UNIVERSE_DOMAIN=googleapis.com | ||
| BUCKET_NAME=dev-sg-dashboard | ||
| CLIENT_ID=1216243614141746 | ||
| # REQUIRED: no default — startup will fail if this is missing |
| # Backward compatible GCS fallback references | ||
| GCS_TYPE=service_account | ||
| GCS_PROJECT_ID=your-gcp-project-id | ||
| GCS_PRIVATE_KEY_ID=your-private-key-id | ||
| GCS_PRIVATE_KEY="-----BEGIN PRIVATE KEY-----\nyour-private-key-here\n-----END PRIVATE KEY-----\n" | ||
| GCS_CLIENT_EMAIL=your-service-account@your-project.iam.gserviceaccount.com | ||
| GCS_CLIENT_ID=your-client-id | ||
| GCS_AUTH_URI=https://accounts.google.com/o/oauth2/auth | ||
| GCS_TOKEN_URI=https://oauth2.googleapis.com/token | ||
| GCS_AUTH_PROVIDER_X509_CERT_URL=https://www.googleapis.com/oauth2/v1/certs | ||
| GCS_CLIENT_X509_CERT_URL=https://www.googleapis.com/robot/v1/metadata/x509/your-service-account-name%40your-project-id.iam.gserviceaccount.com | ||
| GCS_UNIVERSE_DOMAIN=googleapis.com |
There was a problem hiding this comment.
check what is this Backward compatible GCS fallback references
| # Fetch programs / leader categories info from Postgres once for context mapping | ||
| program_info = None | ||
| leader_info = None | ||
| tenant_code = "mitra" | ||
|
|
||
| try: | ||
| async with db.pool.acquire() as conn: | ||
| leader_row = await conn.fetchrow( | ||
| "SELECT id, name, description, tenant_code FROM leader_category WHERE name = $1 LIMIT 1", | ||
| record.get("leader_category") | ||
| ) | ||
| if leader_row: | ||
| leader_info = { | ||
| "id": str(leader_row["id"]), | ||
| "name": leader_row["name"], | ||
| "description": leader_row["description"], | ||
| } | ||
| tenant_code = leader_row["tenant_code"] |
There was a problem hiding this comment.
To get the program and leader IDs first query the database, if not found, then these are new programs, hence generate new uuid's and push then in Kafka
|
|
||
| async def fetchval(self, *args, **kwargs): | ||
| return None | ||
|
|
There was a problem hiding this comment.
check is this is required here
| worker_running = False | ||
|
|
||
|
|
||
| async def run_web_async(): |
There was a problem hiding this comment.
check why this is required
| presidio-anonymizer | ||
| torch | ||
| langdetect | ||
| google-cloud-storage>=2.14.0 |
| cloud_storage_path TEXT NOT NULL, | ||
| meta_data JSONB DEFAULT '{}'::jsonb, | ||
| status VARCHAR(20) NOT NULL DEFAULT 'pending' | ||
| CHECK (status IN ('pending', 'in_progress', 'processed', 'on_hold')), |
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
app/database/operations.py (2)
239-243: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winPrevent
AttributeErrorfrom explicit JSON nulls.If the incoming JSON payload explicitly sets
"newValues": null(or"data": null),event_payload.get("newValues", {})will returnNone, not the default empty dictionary. Consequently,raw_databecomesNone,_clean_val(None)returnsNone, and the subsequentdata.get("submissionDate")on line 253 will crash the worker with anAttributeError.Ensure
raw_dataanddataalways fall back to an empty dictionary to guarantee safe.get()access downstream.🛡️ Proposed safeguard
if event_type == "update": - raw_data = event_payload.get("newValues", {}) + raw_data = event_payload.get("newValues") or {} else: - raw_data = event_payload.get("data", {}) or {} - data = _clean_val(raw_data) + raw_data = event_payload.get("data") or {} + data = _clean_val(raw_data) or {}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/database/operations.py` around lines 239 - 243, Update the raw_data assignment in the event payload handling branch so explicit null values from both “newValues” and “data” fall back to an empty dictionary, then ensure the cleaned data value used by the downstream submissionDate access also defaults to an empty dictionary when _clean_val returns None. Preserve existing payload handling for non-null values.
13-23: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick winOptimize
_clean_vallist comprehension and import overhead.There are two performance concerns in this recursive cleaning function:
- The list comprehension evaluates
_clean_val(item)twice for every element in a list. Because_clean_valis recursive, this double-evaluation results in exponential time complexity for nested lists and doubles the traversal overhead for flat lists containing dictionaries.- Importing
mathinside the function incurs unnecessary overhead on every float encountered, as this function is executed against every node of the JSON payload.Extract the
import mathto the top of the file and use a generator expression (or walrus operator) to evaluate the cleaned item only once.⚡ Proposed optimization
At the top of the file, add
import math. Then apply this fix:- if isinstance(value, float): - import math - if math.isnan(value): - return None - return value - if isinstance(value, str): - if value.lower().strip() in ("nan", "null", "none"): - return None - return value - if isinstance(value, list): - return [_clean_val(item) for item in value if _clean_val(item) is not None] + if isinstance(value, float): + if math.isnan(value): + return None + return value + if isinstance(value, str): + if value.lower().strip() in ("nan", "null", "none"): + return None + return value + if isinstance(value, list): + return [c for c in (_clean_val(item) for item in value) if c is not None]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/database/operations.py` around lines 13 - 23, Optimize _clean_val by moving the math import to module scope and changing the list branch to evaluate _clean_val(item) only once per element, while still filtering out None values and preserving recursive cleaning behavior for nested lists and other values.app/database/db.py (1)
64-66: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick winStop recreating
_connect_lockon every call
app/database/db.py#L64-L66app/database/db.py#L89-L91
getattr(self._connect_lock, '_loop', None)is not a reliable ownership check here, so this condition recreates the lock instead of reusing it. That drops mutual exclusion between concurrentconnect()/disconnect()calls and can race pool setup/teardown. Store the loop explicitly or keep a single lock perDatabaseinstance.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/database/db.py` around lines 64 - 66, Stop recreating _connect_lock in both app/database/db.py ranges 64-66 and 89-91: update the Database lock ownership logic so connect() and disconnect() reuse one stable lock per Database instance, or track its owning loop explicitly instead of inspecting the private _loop attribute. Preserve mutual exclusion across concurrent pool setup and teardown.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@app/temporal/pii_and_abusive_activity.py`:
- Around line 215-218: Fix the indentation of the XML-tag cleanup statements
following the raise ValueError call so they align with the surrounding control
flow and the module parses successfully. Move the re import to the file-level
imports and keep the masked_text normalization in its existing processing path.
---
Outside diff comments:
In `@app/database/db.py`:
- Around line 64-66: Stop recreating _connect_lock in both app/database/db.py
ranges 64-66 and 89-91: update the Database lock ownership logic so connect()
and disconnect() reuse one stable lock per Database instance, or track its
owning loop explicitly instead of inspecting the private _loop attribute.
Preserve mutual exclusion across concurrent pool setup and teardown.
In `@app/database/operations.py`:
- Around line 239-243: Update the raw_data assignment in the event payload
handling branch so explicit null values from both “newValues” and “data” fall
back to an empty dictionary, then ensure the cleaned data value used by the
downstream submissionDate access also defaults to an empty dictionary when
_clean_val returns None. Preserve existing payload handling for non-null values.
- Around line 13-23: Optimize _clean_val by moving the math import to module
scope and changing the list branch to evaluate _clean_val(item) only once per
element, while still filtering out None values and preserving recursive cleaning
behavior for nested lists and other values.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: b79d396b-9c8d-4d44-86e7-f5b25d9e38c8
📒 Files selected for processing (9)
.env.exampleapp/config.pyapp/database/db.pyapp/database/operations.pyapp/temporal/pii_and_abusive_activity.pyapp/temporal/worker.pyapp/temporal/workflows.pyschema.sqlseed_prompts.sql
🚧 Files skipped from review as they are similar to previous changes (5)
- schema.sql
- seed_prompts.sql
- app/temporal/worker.py
- app/config.py
- app/temporal/workflows.py
There was a problem hiding this comment.
Actionable comments posted: 10
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
app/temporal/csv_processing_activity.py (4)
37-44: 🚀 Performance & Scalability | 🔴 Critical | ⚡ Quick winRemove blocking
.get()to enable batching and prevent event loop starvation.Calling
future.get(timeout=10)for every single row defeats Kafka's batching (reducing throughput to one message per network RTT). Furthermore, because_push_rowis executed inside anasync defTemporal activity, this synchronous blocking call will starve the worker's asyncio event loop, preventing other concurrent activities and Temporal heartbeats from running.Remove
.get()and return theFutureso the caller can handle batching and error verification safely.⚡ Proposed fix for batching
-def _push_row(payload: str, key: str | None = None) -> None: +def _push_row(payload: str, key: str | None = None): producer = _get_producer() - future = producer.send( + return producer.send( settings.KAFKA_TOPIC_INGESTION, value=payload, key=key.encode("utf-8") if key else None, ) - future.get(timeout=10)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/csv_processing_activity.py` around lines 37 - 44, Update _push_row to remove the blocking future.get(timeout=10) call and return the Future from producer.send instead. Adjust its return annotation and any callers so they can retain the Future for batching and perform error verification without blocking the async Temporal activity event loop.
209-215: 🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick winCollect Kafka futures and verify them to prevent silent data loss.
If
_push_rowis updated to remove the blocking.get()call to enable batching, you must collect the returned futures and manually verify them after_flush_producer().kafka-python'sflush()does not automatically raise exceptions from failed background requests, so unverified futures will result in silently lost rows while the activity reports success.🛡️ Proposed fix to verify futures
try: + futures = [] for chunk in chunks: for payload in rows_to_json(chunk, report_type, metadata=metadata): record_number += 1 - _push_row(payload, key=f"{record_id}-{pushed}") + futures.append(_push_row(payload, key=f"{record_id}-{pushed}")) pushed += 1 _flush_producer() + for fut in futures: + fut.get(timeout=10)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/csv_processing_activity.py` around lines 209 - 215, Update the row-publishing loop in the activity around _push_row to collect each returned Kafka future, then call _flush_producer() and verify every future afterward so asynchronous send failures are raised instead of being silently ignored. Preserve the existing record numbering, keys, and push count behavior.
67-76: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winDo not swallow transient I/O exceptions; let Temporal handle retries.
Catching all exceptions during the GCS fetch and immediately returning
Falsesubverts Temporal's built-in retry mechanism. A transient network blip will prematurely mark the upload ason_holdwithout any retries, forcing the user to restart the upload process.Only deterministic errors (like column validation failures) should return
Falseimmediately. For transient fetching errors, allow the exception to propagate so Temporal can apply the configuredRetryPolicy.⚡ Proposed fix to allow retries
except Exception as exc: logger.exception("Failed to fetch/load CSV for record %s", record_id) - error_meta = { - "stage": "CSV Fetching", - "error": "Failed to fetch/load CSV from GCS", - "exception": str(exc), - "timestamp": datetime.utcnow().isoformat() + "Z" - } - await csv_upload_repo.update_status(record_id, "on_hold", error_meta) - return False + raise exc🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/csv_processing_activity.py` around lines 67 - 76, Update the CSV fetch/load error handling in the activity’s GCS processing flow to stop catching all exceptions and returning False. Preserve immediate False returns for deterministic validation failures, but allow transient I/O or fetch exceptions to propagate so Temporal’s RetryPolicy can retry them; do not mark those failures on_hold in csv_upload_repo.update_status.
242-248: 🗄️ Data Integrity & Integration | 🔴 Critical | 🏗️ Heavy liftAdd recoverable database-level claiming logic to prevent duplicate processing (TOCTOU race condition).
Fetching pending records without explicitly claiming them creates a critical race condition. If multiple batch processing workflows are inadvertently started (e.g., through overlapping schedules), they will fetch the same pending CSV uploads concurrently. Because downstream activities process these records without atomically verifying their status, this will result in duplicate CSV processing and double ingestion into Kafka.
Based on learnings, you must implement database-level claiming logic. Refactor this activity to perform an atomic update (e.g.,
UPDATE csv_uploads SET status = 'processing' WHERE status = 'pending' RETURNING id) so that only successfully claimed records are returned and processed by the workflow.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/csv_processing_activity.py` around lines 242 - 248, Update fetch_pending_csv_uploads_activity to atomically claim pending csv_upload records by transitioning their status from "pending" to "processing" and returning only the IDs from that update. Add or reuse a repository operation that performs the status-guarded UPDATE with RETURNING semantics, rather than listing pending records first, so concurrent workflows cannot claim the same uploads.Source: Learnings
🧹 Nitpick comments (6)
app/temporal/pii_and_abusive_activity.py (1)
95-99: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove obsolete and misleading comment about array inputs.
The comment explicitly states that discussion columns are
TEXT[], native lists, and that the prompt operates on a "list-input contract". However, this PR changes these fields to be scalar and successfully removes the statement list handling logic. Leaving this comment in place will actively mislead future maintainers.♻️ Proposed refactor
- # Discussion columns (challenges/solutions) are stored as TEXT[] and come - # back from asyncpg as a native list — embed it directly so the model sees - # a proper nested array (and is asked to return one masked entry per - # statement, per the prompt's list-input contract) rather than a flattened - # string.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/pii_and_abusive_activity.py` around lines 95 - 99, Remove the obsolete multi-line comment above the discussion-column handling in the relevant activity flow, including references to TEXT[], native lists, nested arrays, and list-input contracts. Do not alter the scalar field handling or reintroduce list-processing logic.app/temporal/csv_processing_activity.py (2)
64-66: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid blocking the asyncio event loop with synchronous I/O.
fetch_csv(network I/O) andload_csv(CPU-bound parsing) are synchronous functions. Calling them directly inside anasync defTemporal activity blocks the worker's event loop, stalling other concurrent activities and preventing background tasks (like heartbeats) from executing. Offload these blocking operations to a thread pool.
app/temporal/csv_processing_activity.py#L64-L66: Refactor to useawait asyncio.to_thread(fetch_csv, cloud_storage_path)andawait asyncio.to_thread(load_csv, csv_file).app/temporal/csv_processing_activity.py#L108-L109: Apply the sameasyncio.to_threadwrapping here to avoid blocking the event loop during Kafka push initialization.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/csv_processing_activity.py` around lines 64 - 66, In the CSV processing activity, offload the synchronous fetch_csv network call and load_csv parsing call to asyncio.to_thread before assigning their results. Apply the same asyncio.to_thread wrapping to the synchronous Kafka push initialization at app/temporal/csv_processing_activity.py lines 108-109, preserving the existing call arguments and flow.
210-214: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRow tracking mismatch when rows are skipped.
rows_to_jsonsilently skips incomplete rows before yielding them. As a result,record_numbertracks the count of successfully parsed rows rather than the actual line number in the CSV. If a Kafka push fails, therecord_numberin the error metadata will not match the row number in the original CSV file, making it difficult to trace failures back to the source data.Consider refactoring
rows_to_jsonto yield a tuple of(original_row_index, payload)so the absolute line number can be accurately tracked and logged.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/temporal/csv_processing_activity.py` around lines 210 - 214, Refactor rows_to_json to yield each payload together with its original CSV row index, including accounting for skipped incomplete rows. Update the chunk-processing loop to unpack (original_row_index, payload) and use that index for record_number before calling _push_row, while preserving the existing pushed counter and payload handling.app/api/schemas/submission.py (1)
4-7: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse
Literalfor strict validation of the submission type.As per path instructions for
app/api/**regarding "Request validation using Pydantic", consider typingsubmission_typeasLiteral['story', 'discussion']instead of a genericstr. This leverages Pydantic to automatically enforce valid report types during request parsing.♻️ Proposed refactor
+from typing import Literal from pydantic import BaseModel class ManualTriggerRequest(BaseModel): submission_id: str tenant_code: str - submission_type: str + submission_type: Literal["story", "discussion"]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/api/schemas/submission.py` around lines 4 - 7, Update the ManualTriggerRequest.submission_type field to use a Literal type restricted to "story" and "discussion", adding the required typing import. Preserve the existing submission_id and tenant_code fields while relying on Pydantic to reject unsupported submission types during request parsing.Source: Path instructions
app/config.py (1)
257-278: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueInconsistent fallback in
get_discussion_participants_map. Empty/non-dict input returns{}, but a parse failure returns a hardcoded default map. Sincevalidate_participants_map_jsonalready rejects invalid JSON at startup, theexceptbranch is effectively dead and the two "invalid" cases behave differently. Prefer returning{}consistently (or the default consistently) to avoid silently diverging behavior.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/config.py` around lines 257 - 278, Update get_discussion_participants_map so its JSON parse failure path returns the same result as empty or non-dictionary input. Remove the hardcoded fallback map from the except branch and return {} consistently for invalid values, while preserving the existing valid dictionary parsing behavior.schema.sql (1)
390-390: 🗄️ Data Integrity & Integration | 🔵 Trivial | 💤 Low valueConsider constraining
statusto known values.statusaccepts any 20-char string; the pipeline only usespending/in_progress/on_hold/success. ACHECK (status IN (...))(or enum) prevents typos from silently stranding records outsideclaim_pending_records/list_by_statusfilters.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@schema.sql` at line 390, Constrain the status column definition to the pipeline’s supported values: pending, in_progress, on_hold, and success. Update the status declaration near its existing default to add a CHECK constraint (or equivalent enum), while preserving the NOT NULL requirement and pending default used by claim_pending_records and list_by_status.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In @.env.example:
- Line 20: Rename the environment variable in the example configuration from
CRON_BATCH_SIZE to BATCH_SIZE so it matches the setting defined in app/config.py
and preserves the expected deployment configuration contract.
In `@app/api/controllers/csv_upload_controller.py`:
- Around line 73-74: Update the catch-all handlers in the CSV upload flow and
push_record to log the captured exception server-side, then raise HTTPException
with status 500 and a generic client-safe detail instead of str(exc). Preserve
the existing exception handling behavior while ensuring raw internal error text
is not returned to API consumers.
In `@app/api/deps.py`:
- Around line 10-11: Update the authentication check around
credentials.credentials and settings.AUTH_TOKEN to use secrets.compare_digest
instead of !=, importing the secrets module as needed, while preserving the
existing 401 HTTPException for invalid tokens.
In `@app/api/README.md`:
- Around line 21-31: Align the documented staggered schedules with the
configuration defaults by updating either the schedule values in app/config.py
or the corresponding README flowchart. Ensure CSV_SCHEDULE_CRON_TIME and
BATCH_SCHEDULE_CRON produce distinct execution times matching the intended 9:10
PM and 9:15 PM IST coordination.
In `@app/api/validators/csv_upload.py`:
- Around line 10-16: Update validate_report_type to return the validated
normalized_type instead of None, and adjust its return annotation accordingly so
callers can capture and use the lowercased, stripped report type.
In `@app/config.py`:
- Around line 64-65: Update the DISCUSSION_KAFKA_SCHEMA configuration so empty
discussion arrays from parse_segments(), specifically data.challenges,
data.solutions, and data.participantsData, are accepted during ingestion; move
these fields from required to optional or adjust the corresponding
_validate_ingestion_schema() empty-array handling while preserving required
validation for other fields.
In `@app/database/operations.py`:
- Around line 650-679: The duplicate-upload flow around check_duplicate_file and
insert_upload_record must rely on an atomic database constraint rather than a
SELECT-then-insert pre-check; catch asyncpg.exceptions.UniqueViolationError from
the insert and raise DuplicateFile. In schema.sql lines 381-393, add a UNIQUE
constraint or equivalent supporting index covering program_name,
leader_category, report_type, file_name, and file_size to enforce uniqueness and
optimize lookups.
- Around line 779-790: Update the metadata merge in the record update operation
to coalesce the nullable existing meta_data column to an empty JSONB object
before concatenating $2::jsonb. Preserve the current status update and parameter
handling in the surrounding conn.execute call.
In `@app/temporal/csv_processing_activity.py`:
- Around line 131-135: Update the record_meta parsing in the CSV processing
activity to catch only _json.JSONDecodeError, then validate that the parsed
value is a dictionary before calling record_meta.get("tenant_code"). Fall back
to an empty dictionary for invalid JSON or non-dictionary JSON values,
preserving the existing "mitra" tenant default.
In `@app/temporal/pii_and_abusive_activity.py`:
- Around line 148-151: Move the re import to the module-level imports and define
XML_TAG_CLEANUP_RE there with the existing pattern compiled using re.DOTALL. In
the loop containing masked_text cleanup, remove the local import and replace
re.sub with XML_TAG_CLEANUP_RE.sub, preserving the existing replacement and
string conversion.
---
Outside diff comments:
In `@app/temporal/csv_processing_activity.py`:
- Around line 37-44: Update _push_row to remove the blocking
future.get(timeout=10) call and return the Future from producer.send instead.
Adjust its return annotation and any callers so they can retain the Future for
batching and perform error verification without blocking the async Temporal
activity event loop.
- Around line 209-215: Update the row-publishing loop in the activity around
_push_row to collect each returned Kafka future, then call _flush_producer() and
verify every future afterward so asynchronous send failures are raised instead
of being silently ignored. Preserve the existing record numbering, keys, and
push count behavior.
- Around line 67-76: Update the CSV fetch/load error handling in the activity’s
GCS processing flow to stop catching all exceptions and returning False.
Preserve immediate False returns for deterministic validation failures, but
allow transient I/O or fetch exceptions to propagate so Temporal’s RetryPolicy
can retry them; do not mark those failures on_hold in
csv_upload_repo.update_status.
- Around line 242-248: Update fetch_pending_csv_uploads_activity to atomically
claim pending csv_upload records by transitioning their status from "pending" to
"processing" and returning only the IDs from that update. Add or reuse a
repository operation that performs the status-guarded UPDATE with RETURNING
semantics, rather than listing pending records first, so concurrent workflows
cannot claim the same uploads.
---
Nitpick comments:
In `@app/api/schemas/submission.py`:
- Around line 4-7: Update the ManualTriggerRequest.submission_type field to use
a Literal type restricted to "story" and "discussion", adding the required
typing import. Preserve the existing submission_id and tenant_code fields while
relying on Pydantic to reject unsupported submission types during request
parsing.
In `@app/config.py`:
- Around line 257-278: Update get_discussion_participants_map so its JSON parse
failure path returns the same result as empty or non-dictionary input. Remove
the hardcoded fallback map from the except branch and return {} consistently for
invalid values, while preserving the existing valid dictionary parsing behavior.
In `@app/temporal/csv_processing_activity.py`:
- Around line 64-66: In the CSV processing activity, offload the synchronous
fetch_csv network call and load_csv parsing call to asyncio.to_thread before
assigning their results. Apply the same asyncio.to_thread wrapping to the
synchronous Kafka push initialization at app/temporal/csv_processing_activity.py
lines 108-109, preserving the existing call arguments and flow.
- Around line 210-214: Refactor rows_to_json to yield each payload together with
its original CSV row index, including accounting for skipped incomplete rows.
Update the chunk-processing loop to unpack (original_row_index, payload) and use
that index for record_number before calling _push_row, while preserving the
existing pushed counter and payload handling.
In `@app/temporal/pii_and_abusive_activity.py`:
- Around line 95-99: Remove the obsolete multi-line comment above the
discussion-column handling in the relevant activity flow, including references
to TEXT[], native lists, nested arrays, and list-input contracts. Do not alter
the scalar field handling or reintroduce list-processing logic.
In `@schema.sql`:
- Line 390: Constrain the status column definition to the pipeline’s supported
values: pending, in_progress, on_hold, and success. Update the status
declaration near its existing default to add a CHECK constraint (or equivalent
enum), while preserving the NOT NULL requirement and pending default used by
claim_pending_records and list_by_status.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: db207b89-bbcf-49c4-af96-f3c536cbbc87
📒 Files selected for processing (27)
.env.exampleapp/api/README.mdapp/api/controllers/__init__.pyapp/api/controllers/csv_upload_controller.pyapp/api/controllers/submission_controller.pyapp/api/deps.pyapp/api/exceptions.pyapp/api/routes.pyapp/api/schemas/__init__.pyapp/api/schemas/csv_upload.pyapp/api/schemas/submission.pyapp/api/services/__init__.pyapp/api/services/csv_upload_service.pyapp/api/services/submission_service.pyapp/api/validators/__init__.pyapp/api/validators/csv_upload.pyapp/config.pyapp/database/db.pyapp/database/operations.pyapp/storage/gcs.pyapp/temporal/csv_processing_activity.pyapp/temporal/deface_blur_activity.pyapp/temporal/pii_and_abusive_activity.pyapp/temporal/workflows.pymain.pyrequirements.txtschema.sql
💤 Files with no reviewable changes (2)
- requirements.txt
- app/api/routes.py
🚧 Files skipped from review as they are similar to previous changes (2)
- main.py
- app/temporal/workflows.py
2 participants
Summary by CodeRabbit