Add auth + CSV upload/process API - #5
Conversation
…split/single-container app modes Dockerfile + requirements-prod.txt build the app image (CPU-only torch, trimmed unused deps, embedding model baked in). docker-compose.yaml adds a KRaft Kafka broker and analytics-web/consumer/worker as three containers by default (COMPOSE_PROFILES=split), with run_all.sh + analytics-all as a single-container alternative (COMPOSE_PROFILES=single). Postgres stays host-local via host.docker.internal, same as temporal already did.
…ohan's codebase]
Adopts the sibling project's layered app/api/ structure (routes/services/
validators/models) for a real Bearer-token-protected CSV upload pipeline
(POST /v1/upload/, POST /v1/process/csv/{id}), replacing the old
unauthenticated /api/submissions/trigger and mock /api/bulk/upload.
Extracts the Kafka ingestion-schema validator into
app/services/ingestion_validation.py so the CSV pipeline checks each row
against STORY_KAFKA_SCHEMA/DISCUSSION_KAFKA_SCHEMA before publishing,
not just the consumer after receiving. Stops auto-generating a session ID
when missing from a CSV row so it's caught by that check instead of
silently faked, and rejects bad-column CSVs before any GCS/DB write.
Fixes made during the port: confluent_kafka.Producer instead of a second
Kafka client library; blocking GCS/pandas calls wrapped in
asyncio.to_thread inside Temporal activities; RecordNotPending -> 409
(was inconsistently 400).
WalkthroughAdds an authenticated CSV upload API with validation, GCS storage, database tracking, Temporal orchestration, Kafka publication, batch scheduling, and container runtime configuration. Existing submission normalization and ingestion validation paths are also updated for array-valued fields and shared schema validation. ChangesCSV ingestion pipeline
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant UploadRoute
participant UploadService
participant GCS
participant Database
participant Temporal
participant Kafka
Client->>UploadRoute: Upload CSV with bearer token
UploadRoute->>UploadService: Validate and handle upload
UploadService->>GCS: Store CSV bytes
UploadService->>Database: Create pending upload record
UploadService->>Temporal: Start CSV processing workflow
Temporal->>GCS: Fetch stored CSV
Temporal->>Database: Update processing status
Temporal->>Kafka: Publish valid row events
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 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 |
0f60c2b
into
ELEVATE-Project:release-1.0.0
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
🧹 Nitpick comments (4)
docker-compose.yaml (1)
38-39: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winWait for Temporal readiness, not only container creation.
The list-form dependency starts
temporal-uionce the Temporal container starts, even while its new health check is failing. Usecondition: service_healthy.Proposed fix
depends_on: - - temporal + temporal: + condition: service_healthyAs per path instructions, Docker Compose configuration must include effective health checks.
🤖 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 `@docker-compose.yaml` around lines 38 - 39, Update the temporal-ui depends_on entry to use the mapping form with condition: service_healthy for the temporal service, ensuring startup waits for Temporal’s health check rather than container creation alone.Source: Path instructions
run_all.sh (1)
8-16: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winForward shutdown signals and reap sibling processes.
PID 1 exits on
SIGTERMwithout forwarding it to the web, consumer, and worker processes. Add aTERM/INTtrap that terminates and waits for all child PIDs; this lets the Kafka consumer execute its cleanup path before container shutdown.🤖 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 `@run_all.sh` around lines 8 - 16, Update run_all.sh around the web, consumer, and worker launches to capture all child PIDs and install TERM/INT traps that forward the signal to each child, wait for them to exit, then terminate the supervisor cleanly. Preserve the existing wait -n behavior for propagating a child failure while ensuring sibling processes are reaped during signal-driven shutdown.app/services/gcp_storage.py (1)
67-96: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueConsider caching the GCS client / credentials.
Both
upload_csvandfetch_csvrebuildget_gcp_credentials()and instantiate a freshstorage.Clienton every call. A module-level lazily-initialized client would avoid the repeated credential parsing and client setup. Minor, since GCS round-trip latency dominates.🤖 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 67 - 96, Cache the GCS client and credentials used by upload_csv and fetch_csv instead of recreating them on every call. Add module-level lazy initialization and update both functions to reuse the cached storage.Client while preserving the existing bucket, upload, and download behavior.app/api/exceptions.py (1)
87-89: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low value
DuplicateFilemaps more naturally to 409 Conflict than 400.A duplicate upload is a conflict with existing resource state (same class as
RecordAlreadyProcessing/RecordNotPending, which you return as 409), not a malformed request. Consider returning 409 for consistency.🤖 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/exceptions.py` around lines 87 - 89, Update duplicate_file_handler so DuplicateFile responses use HTTP 409 Conflict instead of 400, while preserving the existing JSON detail content and handler behavior.
🤖 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:
- Around line 31-33: Change the AUTH_TOKEN example to a clearly defined sentinel
value, then update the configuration loading and validation in app/config.py to
reject that sentinel and fail fast when it remains configured. Preserve
acceptance of any non-sentinel, explicitly supplied bearer token.
In `@app/api/routes/uploads.py`:
- Around line 21-27: Introduce a Pydantic request model for the upload metadata
used by upload_report, applying required-value validation, appropriate maximum
lengths, and allowed choices where applicable for program_name, leader_category,
and tenant_code. Update upload_report to receive and use this model instead of
unconstrained form strings, while preserving the existing report_type, file, and
authentication dependencies.
In `@app/api/services/uploads.py`:
- Around line 251-255: Update row_to_json’s PDF payload construction so pdf_urls
includes the required data.pdfUrls.masked field alongside original, using the
ingest-path value expected by both Kafka schemas. Ensure
csv_push_to_kafka_activity no longer rejects valid CSV rows when masked is
intentionally produced later, while preserving the existing original URL
handling.
In `@app/temporal/csv_processing_activity.py`:
- Around line 34-63: The module-level producer used by _push_rows_sync must not
be shared across concurrent activity runs. Create or obtain a producer dedicated
to this batch/activity within _push_rows_sync, and ensure produce, delivery
callbacks, polling, and flush(10) operate only on that producer so unrelated
uploads cannot affect delivery tracking or timeout results.
- Around line 104-105: Update csv_fetch_and_validate_activity so claiming the
upload occurs before batch fan-out: move the in_progress status update into the
pending-fetch step, or gate this activity through try_claim_for_processing and
return without processing when the row is no longer pending. Remove the
unconditional update_status call immediately before returning success, while
preserving normal processing for successfully claimed pending rows.
In `@app/temporal/workflows.py`:
- Around line 388-408: Bound the child-workflow fan-out in the pending-upload
processing flow: mirror BatchProcessingWorkflow’s batch_size-limited fetch and
continue_as_new loop, and/or enforce a concurrency cap before asyncio.gather.
Update the section using fetch_pending_csv_uploads_activity,
CsvProcessingWorkflow.run, and child_tasks so thousands of pending IDs are not
fetched or started simultaneously, while preserving per-upload processing and
result handling.
In `@docker-compose.yaml`:
- Around line 41-157: Uncomment and activate the Kafka, analytics-web,
analytics-consumer, analytics-worker, and analytics-all service definitions in
docker-compose.yaml so the split and single COMPOSE_PROFILES can start their
corresponding containers. Preserve the existing profile assignments,
dependencies, environment settings, ports, volumes, and commands.
In `@Dockerfile`:
- Around line 29-35: Create an unprivileged application user in the Dockerfile,
grant it ownership of logs, downloads, and the model cache directories, then
switch to that user with USER before the ENTRYPOINT so the application runs
without root privileges.
In `@requirements-prod.txt`:
- Around line 1-16: Pin the production dependency graph by adding a generated
lock or constraints file for all direct and transitive dependencies, preferably
including package hashes. Update the Dockerfile installation step to install
using this pinned file instead of resolving requirements-prod.txt directly,
while keeping requirements-prod.txt as the source of declared production
dependencies.
---
Nitpick comments:
In `@app/api/exceptions.py`:
- Around line 87-89: Update duplicate_file_handler so DuplicateFile responses
use HTTP 409 Conflict instead of 400, while preserving the existing JSON detail
content and handler behavior.
In `@app/services/gcp_storage.py`:
- Around line 67-96: Cache the GCS client and credentials used by upload_csv and
fetch_csv instead of recreating them on every call. Add module-level lazy
initialization and update both functions to reuse the cached storage.Client
while preserving the existing bucket, upload, and download behavior.
In `@docker-compose.yaml`:
- Around line 38-39: Update the temporal-ui depends_on entry to use the mapping
form with condition: service_healthy for the temporal service, ensuring startup
waits for Temporal’s health check rather than container creation alone.
In `@run_all.sh`:
- Around line 8-16: Update run_all.sh around the web, consumer, and worker
launches to capture all child PIDs and install TERM/INT traps that forward the
signal to each child, wait for them to exit, then terminate the supervisor
cleanly. Preserve the existing wait -n behavior for propagating a child failure
while ensuring sibling processes are reaped during signal-driven shutdown.
🪄 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: 7932e4a5-fa82-4507-8e94-74a41c849ccf
⛔ Files ignored due to path filters (1)
tests/TEST_CASES.csvis excluded by!**/*.csv,!**/*.csv
📒 Files selected for processing (33)
.dockerignore.env.exampleDockerfileapp/api/bulk.pyapp/api/deps.pyapp/api/exceptions.pyapp/api/models/__init__.pyapp/api/models/uploads.pyapp/api/response.pyapp/api/router.pyapp/api/routes.pyapp/api/routes/__init__.pyapp/api/routes/uploads.pyapp/api/services/__init__.pyapp/api/services/uploads.pyapp/api/validators/__init__.pyapp/api/validators/uploads.pyapp/config.pyapp/database/operations.pyapp/kafka/consumer.pyapp/services/gcp_storage.pyapp/services/ingestion_validation.pyapp/temporal/csv_processing_activity.pyapp/temporal/pii_and_abusive_activity.pyapp/temporal/story_rating_activity.pyapp/temporal/thematic_activity.pyapp/temporal/worker.pyapp/temporal/workflows.pydocker-compose.yamlmain.pyrequirements-prod.txtrun_all.shschema.sql
💤 Files with no reviewable changes (2)
- app/api/bulk.py
- app/api/routes.py
| # API Authentication — single shared Bearer token checked via secrets.compare_digest | ||
| # in app/api/deps.py. Required (no default) — the app will not start without it. | ||
| AUTH_TOKEN=your-secret-bearer-token-here |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Reject the example bearer token at configuration load.
If .env.example is copied unchanged, your-secret-bearer-token-here becomes the live shared credential for both protected endpoints. Use a sentinel value and make app/config.py fail fast when that sentinel is configured, rather than accepting a known token.
🤖 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 @.env.example around lines 31 - 33, Change the AUTH_TOKEN example to a
clearly defined sentinel value, then update the configuration loading and
validation in app/config.py to reject that sentinel and fail fast when it
remains configured. Preserve acceptance of any non-sentinel, explicitly supplied
bearer token.
| async def upload_report( | ||
| report_type: str = Form(...), | ||
| program_name: str = Form(...), | ||
| leader_category: str = Form(...), | ||
| tenant_code: str = Form(default="mitra"), | ||
| file: UploadFile = File(...), | ||
| _token: HTTPAuthorizationCredentials = Depends(verify_auth_token), |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate the route and related upload code.
git ls-files 'app/api/**' | sed -n '1,200p'
printf '\n--- uploads route outline ---\n'
ast-grep outline app/api/routes/uploads.py --view expanded || true
printf '\n--- surrounding lines ---\n'
cat -n app/api/routes/uploads.py | sed -n '1,220p'Repository: ELEVATE-Project/analytics-service
Length of output: 3157
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the upload route and associated schemas/services for validation and response contracts.
for f in \
app/api/routes/uploads.py \
app/schemas/*.py \
app/api/**/uploads*.py \
app/services/*.py
do
[ -e "$f" ] && printf '\n### %s ###\n' "$f" && cat -n "$f" | sed -n '1,240p'
doneRepository: ELEVATE-Project/analytics-service
Length of output: 38846
Add a request model for upload metadata. program_name, leader_category, and tenant_code are still plain form strings, so empty or arbitrarily long values can reach persistence. Use Pydantic constraints for required values, length limits, and any allowed choices.
🧰 Tools
🪛 Ruff (0.15.21)
[warning] 26-26: Do not perform function call File in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
[warning] 27-27: Do not perform function call Depends in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
🤖 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/uploads.py` around lines 21 - 27, Introduce a Pydantic request
model for the upload metadata used by upload_report, applying required-value
validation, appropriate maximum lengths, and allowed choices where applicable
for program_name, leader_category, and tenant_code. Update upload_report to
receive and use this model instead of unconstrained form strings, while
preserving the existing report_type, file, and authentication dependencies.
Source: Path instructions
| pdf_col = "Pdf" if normalized_type == "story" else "PDF Urls" | ||
| original_pdf = get_url_field(get_csv_value(row_dict, expected_cols, pdf_col)) | ||
| pdf_urls = None | ||
| if original_pdf: | ||
| pdf_urls = {"original": original_pdf} |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm both schemas require data.pdfUrls.masked at create-time
rg -n 'data\.pdfUrls\.masked' app/config.py
# Confirm row_to_json never emits a "masked" pdf key
rg -n '"masked"|pdf_urls\s*=' app/api/services/uploads.py
# Confirm pre-publish validation skips failing rows
rg -n 'validate_ingestion_schema|schema_errors\.append|payloads\.append' app/temporal/csv_processing_activity.pyRepository: ELEVATE-Project/analytics-service
Length of output: 2286
CSV create payloads miss data.pdfUrls.masked
row_to_json only emits {"original": ...}, but both STORY_KAFKA_SCHEMA and DISCUSSION_KAFKA_SCHEMA require data.pdfUrls.masked at create time. csv_push_to_kafka_activity rejects failing rows before publish, so CSV uploads never reach Kafka. If masked is produced later, it should not be required in this ingest 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/services/uploads.py` around lines 251 - 255, Update row_to_json’s PDF
payload construction so pdf_urls includes the required data.pdfUrls.masked field
alongside original, using the ingest-path value expected by both Kafka schemas.
Ensure csv_push_to_kafka_activity no longer rejects valid CSV rows when masked
is intentionally produced later, while preserving the existing original URL
handling.
| def _push_rows_sync(payloads: List[Any]) -> None: | ||
| """ | ||
| Runs in a worker thread (via asyncio.to_thread) — produce()/flush() are | ||
| blocking calls. Flushes once for the whole batch rather than per row (a | ||
| per-row flush forces a network round trip per row, far too slow for large | ||
| CSVs), mirroring app/kafka/consumer.py's DLQ producer pattern. | ||
| """ | ||
| producer = _get_producer() | ||
| delivery_error = {} | ||
|
|
||
| def _on_delivery(err, _msg): | ||
| if err is not None: | ||
| delivery_error["error"] = err | ||
|
|
||
| for payload, key in payloads: | ||
| producer.produce( | ||
| settings.KAFKA_TOPIC_INGESTION, | ||
| value=payload.encode("utf-8"), | ||
| key=key.encode("utf-8") if key else None, | ||
| callback=_on_delivery, | ||
| ) | ||
| producer.poll(0) | ||
| if "error" in delivery_error: | ||
| raise KafkaException(delivery_error["error"]) | ||
|
|
||
| remaining = producer.flush(10) | ||
| if remaining > 0: | ||
| raise TimeoutError(f"Timed out waiting for Kafka delivery ({remaining} still in-flight)") | ||
| if "error" in delivery_error: | ||
| raise KafkaException(delivery_error["error"]) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm the producer is a shared singleton and how many concurrent push paths exist.
rg -nP '_producer|_get_producer|_push_rows_sync' app/temporal/csv_processing_activity.py
rg -nP 'execute_child_workflow|asyncio\.gather|csv_push_to_kafka_activity' app/temporal/workflows.pyRepository: ELEVATE-Project/analytics-service
Length of output: 867
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== app/temporal/csv_processing_activity.py (selected lines) ==\n'
sed -n '1,140p' app/temporal/csv_processing_activity.py
printf '\n== app/temporal/workflows.py (selected lines) ==\n'
sed -n '240,430p' app/temporal/workflows.pyRepository: ELEVATE-Project/analytics-service
Length of output: 12632
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== app/temporal/csv_processing_activity.py (push activity slice) ==\n'
sed -n '140,260p' app/temporal/csv_processing_activity.pyRepository: ELEVATE-Project/analytics-service
Length of output: 5292
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== app/temporal/csv_processing_activity.py (remaining push activity) ==\n'
sed -n '260,380p' app/temporal/csv_processing_activity.pyRepository: ELEVATE-Project/analytics-service
Length of output: 1515
Avoid sharing this producer across activity runs. app/temporal/csv_processing_activity.py:20-63 producer.flush(10) drains the entire queue on the module-level singleton, so concurrent uploads on the same worker can block on each other’s in-flight messages and raise a spurious TimeoutError, pushing an otherwise successful record to on_hold. Use a per-activity producer or otherwise isolate delivery tracking to this batch.
🤖 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 34 - 63, The
module-level producer used by _push_rows_sync must not be shared across
concurrent activity runs. Create or obtain a producer dedicated to this
batch/activity within _push_rows_sync, and ensure produce, delivery callbacks,
polling, and flush(10) operate only on that producer so unrelated uploads cannot
affect delivery tracking or timeout results.
| await csv_upload_repo.update_status(record_id, "in_progress") | ||
| return True |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
set -euo pipefail
printf 'Files:\n'
git ls-files 'app/temporal/**' | sed -n '1,200p'
printf '\nRelevant search:\n'
rg -n "try_claim_for_processing|update_status\\(|pending|CsvBatchProcessingWorkflow|handle_push|Kafka|publish" app/temporal -S
printf '\nFile outline:\n'
ast-grep outline app/temporal/csv_processing_activity.py --view expandedRepository: ELEVATE-Project/analytics-service
Length of output: 10488
🏁 Script executed:
set -euo pipefail
sed -n '1,220p' app/temporal/csv_processing_activity.py | cat -n
printf '\n----\n'
sed -n '1,260p' app/temporal/csv_processing_workflow.py | cat -n
printf '\n----\n'
rg -n "try_claim_for_processing|handle_push|update_status\\(|in_progress|pending" app -SRepository: ELEVATE-Project/analytics-service
Length of output: 10488
🏁 Script executed:
set -euo pipefail
printf '--- workflows.py relevant sections ---\n'
sed -n '360,430p' app/temporal/workflows.py | cat -n
printf '\n--- search for claim/update helpers ---\n'
rg -n "try_claim_for_processing|update_status\\(|list_by_status\\(|claim.*processing|in_progress|handle_push" app/database app -S
printf '\n--- database operations file outline ---\n'
fd -a 'operations.py' app/databaseRepository: ELEVATE-Project/analytics-service
Length of output: 262
🏁 Script executed:
set -euo pipefail
printf '--- workflows.py slice ---\n'
sed -n '375,420p' app/temporal/workflows.py | cat -n
printf '\n--- operations.py outline and claim/update functions ---\n'
ast-grep outline app/database/operations.py --view expanded
printf '\n----\n'
rg -n "def try_claim_for_processing|def update_status|list_by_status|pending|in_progress" app/database/operations.py -SRepository: ELEVATE-Project/analytics-service
Length of output: 262
🏁 Script executed:
set -euo pipefail
printf -- '--- workflows.py slice ---\n'
sed -n '375,430p' app/temporal/workflows.py | cat -n
printf -- '\n--- operations.py outline ---\n'
ast-grep outline app/database/operations.py --view expanded
printf -- '\n--- operations.py search ---\n'
rg -n "def try_claim_for_processing|def update_status|def list_by_status|pending|in_progress" app/database/operations.py -SRepository: ELEVATE-Project/analytics-service
Length of output: 4025
🏁 Script executed:
set -euo pipefail
printf -- '--- CsvProcessingWorkflow slice ---\n'
sed -n '300,375p' app/temporal/workflows.py | cat -n
printf -- '\n--- operations.py claim/update slice ---\n'
sed -n '704,792p' app/database/operations.py | cat -nRepository: ELEVATE-Project/analytics-service
Length of output: 6899
🏁 Script executed:
set -euo pipefail
printf -- '--- usages of CsvProcessingWorkflow / try_claim_for_processing / fetch_pending_csv_uploads_activity ---\n'
rg -n "CsvProcessingWorkflow|try_claim_for_processing|fetch_pending_csv_uploads_activity|csv_fetch_and_validate_activity|csv_push_to_kafka_activity" app -S
printf -- '\n--- find any direct single-record CSV processing entrypoints ---\n'
rg -n "csv_uploads.*pending|status = 'pending'|csv_upload|handle_push|push.*csv|process.*csv" app -SRepository: ELEVATE-Project/analytics-service
Length of output: 6563
Claim CSV uploads before batch fan-out.
csv_fetch_and_validate_activity still does an unconditional update_status(..., "in_progress"), so the batch workflow can start on a row that handle_push has already claimed and process it twice. Move the claim into the pending-fetch step or gate this activity on try_claim_for_processing, and skip rows that are no longer pending.
🤖 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 104 - 105, Update
csv_fetch_and_validate_activity so claiming the upload occurs before batch
fan-out: move the in_progress status update into the pending-fetch step, or gate
this activity through try_claim_for_processing and return without processing
when the row is no longer pending. Remove the unconditional update_status call
immediately before returning success, while preserving normal processing for
successfully claimed pending rows.
Source: Learnings
| pending_ids: List[int] = await workflow.execute_activity( | ||
| fetch_pending_csv_uploads_activity, | ||
| start_to_close_timeout=timedelta(minutes=2), | ||
| retry_policy=retry_policy | ||
| ) | ||
|
|
||
| if not pending_ids: | ||
| return {"processed_count": 0, "message": "No pending CSV uploads found."} | ||
|
|
||
| # Fan-out child workflows to process each CSV in parallel | ||
| child_tasks = [] | ||
| for pid in pending_ids: | ||
| child_tasks.append( | ||
| workflow.execute_child_workflow( | ||
| CsvProcessingWorkflow.run, | ||
| pid, | ||
| id=f"csv-batch-child-{pid}" | ||
| ) | ||
| ) | ||
|
|
||
| results = await asyncio.gather(*child_tasks, return_exceptions=True) |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
Unbounded fan-out diverges from BatchProcessingWorkflow's deliberate bounding.
fetch_pending_csv_uploads_activity returns all pending IDs, and every one is launched as a child workflow in a single asyncio.gather. The sibling BatchProcessingWorkflow in this same file explicitly fetches in batch_size chunks and uses continue_as_new precisely to avoid OOM on the worker and overwhelming the Temporal cluster with concurrent starts. At a few thousand pending uploads this workflow hits exactly the failure mode that pattern was written to prevent.
Recommend bounding the pending fetch (limit + loop, mirroring BatchProcessingWorkflow) and/or capping concurrent children.
🤖 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 388 - 408, Bound the child-workflow
fan-out in the pending-upload processing flow: mirror BatchProcessingWorkflow’s
batch_size-limited fetch and continue_as_new loop, and/or enforce a concurrency
cap before asyncio.gather. Update the section using
fetch_pending_csv_uploads_activity, CsvProcessingWorkflow.run, and child_tasks
so thousands of pending IDs are not fetched or started simultaneously, while
preserving per-upload processing and result handling.
| # kafka: | ||
| # image: apache/kafka:3.9.0 | ||
| # container_name: kafka | ||
| # ports: | ||
| # - "29092:29092" # host-facing listener only — do NOT also publish 9092 | ||
| # environment: | ||
| # KAFKA_NODE_ID: 1 | ||
| # KAFKA_PROCESS_ROLES: broker,controller | ||
| # KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,PLAINTEXT_HOST://:29092 | ||
| # KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092 | ||
| # KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT | ||
| # KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER | ||
| # KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT | ||
| # KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 | ||
| # KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 | ||
| # KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 | ||
| # CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk" | ||
| # volumes: | ||
| # - kafka_data:/var/lib/kafka/data | ||
| # healthcheck: | ||
| # test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server kafka:9092 || exit 1"] | ||
| # interval: 10s | ||
| # timeout: 5s | ||
| # retries: 10 | ||
| # start_period: 20s | ||
|
|
||
| # # --- Split mode: one container per service (default — fault isolation, | ||
| # # independent scaling/restarts). Active when COMPOSE_PROFILES=split (see .env). | ||
| # analytics-web: | ||
| # profiles: ["split"] | ||
| # build: . | ||
| # image: elevate-analytics:latest | ||
| # env_file: .env | ||
| # environment: | ||
| # DATABASE_URL: postgresql://postgres:[email protected]:5432/analytics_db | ||
| # TEMPORAL_HOST: temporal:7233 | ||
| # KAFKA_BOOTSTRAP_SERVERS: kafka:9092 | ||
| # extra_hosts: | ||
| # - "host.docker.internal:host-gateway" | ||
| # depends_on: | ||
| # temporal: | ||
| # condition: service_healthy | ||
| # kafka: | ||
| # condition: service_healthy | ||
| # ports: | ||
| # - "8000:8000" | ||
| # volumes: | ||
| # - ./logs:/app/logs | ||
| # command: ["--mode", "web"] | ||
|
|
||
| # analytics-consumer: | ||
| # profiles: ["split"] | ||
| # build: . | ||
| # image: elevate-analytics:latest | ||
| # env_file: .env | ||
| # environment: | ||
| # DATABASE_URL: postgresql://postgres:[email protected]:5432/analytics_db | ||
| # TEMPORAL_HOST: temporal:7233 | ||
| # KAFKA_BOOTSTRAP_SERVERS: kafka:9092 | ||
| # extra_hosts: | ||
| # - "host.docker.internal:host-gateway" | ||
| # depends_on: | ||
| # temporal: | ||
| # condition: service_healthy | ||
| # kafka: | ||
| # condition: service_healthy | ||
| # volumes: | ||
| # - ./logs:/app/logs | ||
| # command: ["--mode", "consumer"] | ||
|
|
||
| # analytics-worker: | ||
| # profiles: ["split"] | ||
| # build: . | ||
| # image: elevate-analytics:latest | ||
| # env_file: .env | ||
| # environment: | ||
| # DATABASE_URL: postgresql://postgres:[email protected]:5432/analytics_db | ||
| # TEMPORAL_HOST: temporal:7233 | ||
| # KAFKA_BOOTSTRAP_SERVERS: kafka:9092 | ||
| # extra_hosts: | ||
| # - "host.docker.internal:host-gateway" | ||
| # depends_on: | ||
| # temporal: | ||
| # condition: service_healthy | ||
| # kafka: | ||
| # condition: service_healthy | ||
| # volumes: | ||
| # - ./logs:/app/logs | ||
| # command: ["--mode", "worker"] | ||
|
|
||
| # --- Single mode: one container, all three services as separate processes | ||
| # (run_all.sh). Active when COMPOSE_PROFILES=single (see .env). Do not run | ||
| # this alongside the split services above — both bind host port 8000. | ||
| # analytics-all: | ||
| # profiles: ["single"] | ||
| # build: . | ||
| # image: elevate-analytics:latest | ||
| # env_file: .env | ||
| # environment: | ||
| # DATABASE_URL: postgresql://postgres:[email protected]:5432/analytics_db | ||
| # TEMPORAL_HOST: temporal:7233 | ||
| # KAFKA_BOOTSTRAP_SERVERS: kafka:9092 | ||
| # extra_hosts: | ||
| # - "host.docker.internal:host-gateway" | ||
| # depends_on: | ||
| # temporal: | ||
| # condition: service_healthy | ||
| # kafka: | ||
| # condition: service_healthy | ||
| # ports: | ||
| # - "8000:8000" | ||
| # volumes: | ||
| # - ./logs:/app/logs | ||
| # entrypoint: ["/app/run_all.sh"] | ||
|
|
||
| volumes: | ||
| kafka_data: |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
docker compose --profile split config --services
docker compose --profile single config --servicesRepository: ELEVATE-Project/analytics-service
Length of output: 255
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== docker-compose.yaml (outline) ==\n'
wc -l docker-compose.yaml
sed -n '1,240p' docker-compose.yaml | cat -n
printf '\n== profile references ==\n'
rg -n "COMPOSE_PROFILES|profiles:|analytics-web|analytics-consumer|analytics-worker|analytics-all|kafka:" -S .Repository: ELEVATE-Project/analytics-service
Length of output: 8557
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Show the compose file with line numbers around the referenced region and the file start.
sed -n '1,220p' docker-compose.yaml | nl -ba | sed -n '1,220p'Repository: ELEVATE-Project/analytics-service
Length of output: 211
Expose the profiled services in Compose. COMPOSE_PROFILES=split/single won’t add Kafka or any analytics container while these blocks stay commented out, so the stack only starts temporal and temporal-ui. Make these service definitions active here or move them to docs.
🤖 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 `@docker-compose.yaml` around lines 41 - 157, Uncomment and activate the Kafka,
analytics-web, analytics-consumer, analytics-worker, and analytics-all service
definitions in docker-compose.yaml so the split and single COMPOSE_PROFILES can
start their corresponding containers. Preserve the existing profile assignments,
dependencies, environment settings, ports, volumes, and commands.
Source: Path instructions
| COPY app/ app/ | ||
| COPY main.py schema.sql seed_prompts.sql seed_themes.sql run_all.sh ./ | ||
| RUN chmod +x run_all.sh | ||
|
|
||
| RUN mkdir -p logs downloads | ||
|
|
||
| ENTRYPOINT ["python", "main.py"] |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Run the application as a non-root user.
The entrypoint runs as root because the image never switches users. Create an unprivileged user, grant it ownership of writable directories (logs, downloads, and the model cache), then add USER.
Proposed fix
RUN mkdir -p logs downloads
+RUN useradd --system --create-home --uid 10001 appuser \
+ && chown -R appuser:appuser /app /opt/model-cache
+
+USER appuser
ENTRYPOINT ["python", "main.py"]🤖 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 `@Dockerfile` around lines 29 - 35, Create an unprivileged application user in
the Dockerfile, grant it ownership of logs, downloads, and the model cache
directories, then switch to that user with USER before the ENTRYPOINT so the
application runs without root privileges.
Source: Linters/SAST tools
| temporalio>=1.18.0 | ||
| confluent-kafka>=2.4.0 | ||
| asyncpg>=0.29.0 | ||
| pydantic-settings>=2.0.0 | ||
| deface>=1.5.0 | ||
| python-dotenv>=1.0.1 | ||
| fastapi>=0.112.0 | ||
| python-multipart>=0.0.9 | ||
| uvicorn>=0.30.0 | ||
| sentence-transformers==3.0.1 | ||
| scikit-learn | ||
| torch | ||
| numpy>=1.24.0 | ||
| pandas>=2.0.0 | ||
| google-cloud-storage | ||
| pypdf>=4.0.0 |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== files ==\n'
git ls-files | rg '(^|/)(requirements-prod\.txt|requirements.*\.txt|Dockerfile|dockerfile|constraints.*\.txt|poetry\.lock|uv\.lock|pip-tools|Pipfile\.lock|requirements.*lock.*)$'
printf '\n== requirements-prod.txt ==\n'
cat -n requirements-prod.txt
printf '\n== Dockerfile references ==\n'
rg -n --hidden --no-ignore-vcs 'requirements-prod\.txt|pip install|constraints|requirements' Dockerfile . -g '!**/.git/**' || trueRepository: ELEVATE-Project/analytics-service
Length of output: 50390
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'requirements-prod.txt:\n'
cat -n requirements-prod.txt
printf '\nTracked Dockerfiles:\n'
git ls-files | rg '(^|/)(Dockerfile|dockerfile)$'
printf '\nRequirements usage:\n'
rg -n --hidden --no-ignore-vcs 'requirements-prod\.txt|pip install|constraints|requirements' -- Dockerfile . -g '!**/.git/**' -g '!**/node_modules/**' -g '!**/dist/**' -g '!**/build/**' | head -n 200Repository: ELEVATE-Project/analytics-service
Length of output: 50390
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '== tracked packaging files ==\n'
git ls-files -- 'requirements-prod.txt' 'Dockerfile' 'dockerfile' 'requirements*.txt' 'constraints*.txt' '*.lock'
printf '\n== requirements-prod.txt ==\n'
cat -n requirements-prod.txt
for f in Dockerfile dockerfile; do
if [ -f "$f" ]; then
printf '\n== %s ==\n' "$f"
cat -n "$f"
fi
doneRepository: ELEVATE-Project/analytics-service
Length of output: 2296
Lock the production dependency graph
requirements-prod.txt uses mostly open-ended lower bounds, and Dockerfile installs it directly. Add a pinned lock/constraints file, ideally with hashes, and install from that so rebuilds don’t drift to different or vulnerable transitive versions.
🤖 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 `@requirements-prod.txt` around lines 1 - 16, Pin the production dependency
graph by adding a generated lock or constraints file for all direct and
transitive dependencies, preferably including package hashes. Update the
Dockerfile installation step to install using this pinned file instead of
resolving requirements-prod.txt directly, while keeping requirements-prod.txt as
the source of declared production dependencies.
1 participant
Summary by CodeRabbit
New Features
Bug Fixes