add temporal service#25
Conversation
📝 WalkthroughWalkthroughThis PR introduces Temporal workflow orchestration for table profiling, replacing/augmenting local Trino-based profiling with a new worker service. It adds an ChangesTemporal Profiling and Partial Resume
Estimated code review effort: 4 (Complex) | ~75 minutes Sequence Diagram(s)sequenceDiagram
participant Frontend
participant ProfilingRouter
participant Temporal
participant Worker as TableProfilingWorkflow
Frontend->>ProfilingRouter: POST /profile/run (resume_from_partial)
ProfilingRouter->>Temporal: start workflow
Temporal->>Worker: run(table_id, resume_from_partial)
Worker->>Worker: fetch metadata, compute metrics, profile columns
Worker->>Worker: generate AI summary, persist results
Frontend->>ProfilingRouter: GET /profile (poll)
ProfilingRouter->>Temporal: query workflow status
Temporal-->>ProfilingRouter: status
ProfilingRouter-->>Frontend: TableProfile (status, is_partial)
Frontend->>ProfilingRouter: POST /profile/terminate
ProfilingRouter->>Temporal: cancel workflow
ProfilingRouter-->>Frontend: profile marked failed
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 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 |
| logger.info("[Profiling] Successfully started Temporal workflow for table %s", table_id) | ||
| return True | ||
| except Exception as e: | ||
| if "WorkflowAlreadyStartedError" in str(type(e)) or "WorkflowExecutionAlreadyStartedError" in str(type(e)): |
There was a problem hiding this comment.
Can't you just check the type like
except (WorkflowAlreadyStartedError, WorkflowExecutionAlreadyStartedError):
| return False | ||
|
|
||
|
|
||
| def _run_profile_job_local(table_id: str): |
There was a problem hiding this comment.
I think if we set it to run at temporal, we dont want option to run that local, just throw an error so the user know that currently can't do profiling.
If you want local lets do it like we did in the esca with the env LOCAL_PROFILE (but not sure we really want profiling to be on backend anymore)
|
@coderabbitai review |
✅ Action performedReview finished.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 22
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (5)
worker/app/workflows/profiling_workflow.py (1)
25-213: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider extracting helper methods to reduce workflow complexity.
Static analysis flags this method for excessive branches/statements (PLR0912/PLR0915). Splitting the chunk-profiling, column-profiling, and summary-generation steps into private workflow methods would improve readability without changing 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 `@worker/app/workflows/profiling_workflow.py` around lines 25 - 213, The run method in TableProfilingWorkflow is too complex and should be split into smaller private helpers to satisfy the PLR0912/PLR0915 warning. Extract the chunk-profiling logic, per-column profiling execution, and AI summary generation into dedicated methods on the same workflow class, then have run orchestrate those helpers while preserving the existing flow and error handling.Source: Linters/SAST tools
backend/app/infra_init.py (1)
1159-1241: 📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy liftExtract shared profile-persistence logic to avoid divergence.
This
_run_profileclosure duplicates almost the entire persistence block (status/is_partial computation, field copy,ColumnProfileconstruction) found inbackend/app/routers/profiling.py::_run_profile_job_local. The duplication has already caused a behavioral divergence: hereprofile.statusis set conditionally onresult.success, while the router's copy hardcodesProfilingStatus.completedunconditionally (see comment onprofiling.pylines 136-147). Consolidating into a single shared helper (e.g., incore.services.profiling_engine) would prevent this class of inconsistency going forward.🤖 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 `@backend/app/infra_init.py` around lines 1159 - 1241, The profile persistence logic in `_run_profile` is duplicated and has already drifted from `routers/profiling.py::_run_profile_job_local`, so extract the shared status/field-copy/`ColumnProfile` save flow into one helper and call it from both places. Move the common persistence steps (including the `ProfilingStatus` selection, `is_partial`, profile field assignments, and `ColumnProfile` creation) into a shared service such as `core.services.profiling_engine`, then update `_run_profile` to use that helper so both code paths stay consistent.backend/app/routers/profiling.py (2)
136-147: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winFailed local runs are persisted as "completed".
The prior early-return-on-failure was removed (per the line-range summary), but
profile.status = ProfilingStatus.completedon line 146 is unconditional — it no longer branches onresult.success. A total profiling failure (e.g.,row_count/sample_dataallNone, no column stats) now gets written asstatus="completed", is_partial=Trueinstead ofstatus="failed".This diverges from the equivalent path in
backend/app/infra_init.py::_run_profile(lines 1190-1195), which still conditionally setscompletedvsfailed. On the frontend,ProfilingTab.tsxgates its "Profiling job failed" banner strictly onprofile?.status === 'failed'and the main detail view onprofile?.row_count !== undefined— if both are false/undefined simultaneously (failure with no row_count, but status says "completed"), the UI renders nothing informative for that state.🐛 Proposed fix
- profile.status = ProfilingStatus.completed + profile.status = ( + ProfilingStatus.completed if result.success else ProfilingStatus.failed + ) profile.is_partial = not result.success or bool(result.errors)🤖 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 `@backend/app/routers/profiling.py` around lines 136 - 147, The profiling persistence path in the table profile upsert always marks results as completed, even when the run failed. Update the status assignment in the Session/TableProfile save block to branch on result.success so total failures are stored as ProfilingStatus.failed and only successful runs as completed, while keeping is_partial based on partial results/errors. Align this logic with the existing conditional status handling in _run_profile in infra_init.py and use TableProfile.status / TableProfile.is_partial as the main fields to update.
261-283: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winWire
forcethrough the profiling endpointsBoth
run_all_profilesandrun_table_profileacceptforce, but neither passes it into_trigger_profilingor uses it to bypass the 24h cache. The frontend’sforce: truepath is currently ignored, so either thread it through the workflow/local fallback or remove the parameter and update the API/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 `@backend/app/routers/profiling.py` around lines 261 - 283, The profiling endpoints accept a force flag but run_all_profiles currently drops it when scheduling _trigger_profiling, so the frontend’s force=true request is ignored. Thread force through the background task call and into the profiling flow used by run_table_profile/_trigger_profiling so it can bypass the 24h cache, or remove the unused parameter from both endpoints and align the API/docs accordingly.core/src/core/services/profiling_engine.py (1)
691-753: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚖️ Poor tradeoff
_process_column_metricscomplexity flagged by static analysis (16 branches > 12).The precomputed-vs-fresh-query branching combined with per-semantic-type extraction makes this function hard to follow/extend. Consider extracting a small per-semantic-type dispatch (dict of handlers) to separate the "fetch nulls/distinct" concern from the "extract type-specific stats" concern.
🤖 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 `@core/src/core/services/profiling_engine.py` around lines 691 - 753, The `_process_column_metrics` function is too branch-heavy because it mixes null/distinct fetching with semantic-type-specific extraction. Refactor it by extracting the shared precomputed-vs-query logic from `_process_column_metrics`, then replace the long `if/elif` chain with a small dispatch table or helper handlers for `"categorical"`, `"boolean"`, `"large_categorical"`, `"continuous"`, and `"time"`. Keep the existing helpers like `_extract_categorical_stats`, `_extract_continuous_stats`, `_extract_time_stats`, and the top-values query path, but move each type-specific block into its own function to simplify the main flow.Source: Linters/SAST tools
♻️ Duplicate comments (2)
backend/app/routers/profiling.py (2)
106-111: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy liftLocal Temporal fallback still present; unresolved from prior review.
_run_profile_job_localremains a silent fallback whenever Temporal fails to start, and_trigger_profilinginvokes it without surfacing any error to the caller. This matches an unresolved concern from a prior review round about not silently falling back to local profiling once Temporal is the intended execution path.Separately,
resume_from_partialis accepted by_trigger_profilingbut is dropped entirely when falling back to_run_profile_job_local(line 212 only passestable_id), so a user-requested "resume" silently becomes a full local re-run if Temporal is unavailable.As per past review comment: I think if we set it to run at temporal, we dont want option to run that local, just throw an error so the user know that currently can't do profiling.
Also applies to: 208-212
🤖 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 `@backend/app/routers/profiling.py` around lines 106 - 111, The profiling flow still silently falls back from Temporal to local execution, which should be removed; update `_trigger_profiling` so that if Temporal cannot start, it raises an error instead of calling `_run_profile_job_local`. Also ensure `resume_from_partial` is not accepted as a no-op in this fallback path—keep the Temporal-only behavior explicit in `_trigger_profiling` and `_run_profile_job_local` so users are told profiling cannot proceed when Temporal is unavailable.
88-103: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winFragile exception matching; still unresolved from prior review.
"WorkflowAlreadyStartedError" in str(type(e))matches by stringifying the exception's type rather than catching the actual exception class, which is brittle (breaks on wrapping/renaming, subclassing, or module path changes) and was already flagged in a prior review round.Also,
Client.connect(settings.TEMPORAL_HOST)is invoked here on every trigger call with no client reuse; the same pattern repeats inget_table_profile(line 234) andterminate_table_profile(line 340), andget_table_profileis polled every 2s from the frontend while a profile is running/pending — creating a fresh gRPC connection per poll. Consider connecting once at app startup (e.g., via FastAPI lifespan) and reusing the client.♻️ Example fix for exception handling
- except Exception as e: - if "WorkflowAlreadyStartedError" in str(type(e)) or "WorkflowExecutionAlreadyStartedError" in str(type(e)): + except (WorkflowAlreadyStartedError, WorkflowExecutionAlreadyStartedError): + logger.info("[Profiling] Profiling workflow is already running for table %s", table_id) + return True + except Exception as e: logger.info("[Profiling] Profiling workflow is already running for table %s", table_id) return True - logger.error("[Profiling] Failed to start Temporal workflow for table %s: %s", table_id, e) - return FalseAs per past review comment: Can't you just check the type like
except (WorkflowAlreadyStartedError, WorkflowExecutionAlreadyStartedError):🤖 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 `@backend/app/routers/profiling.py` around lines 88 - 103, The Temporal workflow start handling in profiling should stop matching exceptions by stringifying e.g. in the workflow trigger path and instead catch the actual Temporal exception classes directly. Update the exception handling around Client.start_workflow in the profiling router to use explicit exception types such as WorkflowAlreadyStartedError and WorkflowExecutionAlreadyStartedError, and keep the existing “already running” branch for those cases. Also reuse a shared Temporal client instead of calling Client.connect in each of the profiling entry points, including get_table_profile and terminate_table_profile, ideally by initializing it once at app startup and injecting it where needed.
🤖 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 `@backend/app/routers/profiling.py`:
- Around line 231-255: The Temporal sync block in the profiling router is
checking the wrong exception shape, so missing workflows are not being detected
and stuck profiles remain running or pending. Update the try/except around
Client.connect, get_workflow_handle, and handle.describe to catch the RPCError
from Temporal and branch on its status code for NOT_FOUND, then mark the profile
as failed and persist it; keep the existing failure handling for other workflow
terminal states and log unexpected errors through the existing logger.warning
path.
- Around line 326-356: The terminate_table_profile endpoint is swallowing
cancellation failures and always marking the TableProfile as failed, even when
Client.connect/get_workflow_handle/handle.cancel does not succeed. Update
terminate_table_profile to surface the exception path from the Temporal cancel
attempt (instead of continuing to success), and only set profile.status to
failed and return the terminated response after a confirmed cancel. Keep the
flow tied to the existing terminate_table_profile, TableProfile, and
handle.cancel logic so the API does not report success when the workflow may
still be running.
In `@backend/pyproject.toml`:
- Line 28: The backend dependency list currently allows temporalio to float with
only a lower bound, which risks untested upgrades breaking the worker. Update
the dependency entry in pyproject.toml for temporalio to use an exact pin or a
bounded version range consistent with other locked dependencies like sqlmodel,
so future installs remain predictable.
In `@core/pyproject.toml`:
- Around line 15-16: The dependency entries for langchain and langchain-openai
are unpinned, which can lead to non-reproducible builds and unexpected breakage.
Update the dependency definitions in pyproject.toml to use a known-compatible
version range or exact pins, matching the project’s existing pinned dependency
style like sqlmodel==0.0.22, so the package resolver always installs the
intended versions.
In `@core/src/core/services/profiling_engine.py`:
- Around line 336-376: Preserve the original column name when building
combined-result lookup keys in parse_combined_result, because lowercasing
col_name causes mixed-case or quoted aliases to miss the __non_null and
__distinct entries and trigger per-column fallbacks. Update the key construction
in parse_combined_result to use the original column name for row_dict lookups
while keeping any case-insensitive handling only where truly needed, and verify
the metrics map still uses the original col_name as the output key.
- Around line 316-328: The time-type handling in profiling_engine’s
combined-query stats block is using an exact match that misses parameterized
Trino timestamp types like timestamp(3). Update the time branch in the profiling
logic that builds select_parts to use the same substring-based check as
_classify_semantic_type, so all TIME_TYPES variants are recognized and
time-specific aggregates are generated consistently.
- Around line 1024-1034: Guard the LLM invocation in the profiling flow so it is
skipped when no API key is configured. In the code that builds and uses
ChatOpenAI in profiling_engine, only pass api_key when settings.LLM_API_KEY is
set (or return early / use a fallback when it is unset), so invoke() is never
called with api_key=None. Keep the fix localized to the ChatOpenAI construction
and response = llm.invoke(prompt) path.
In `@docker-compose.yml`:
- Around line 415-430: The docker-compose service definition for temporal-ui
uses an unpinned image tag, which is inconsistent with the pinned temporal
service image and can cause non-reproducible deployments. Update the temporal-ui
image reference in the compose file to a specific version tag that matches your
intended release strategy, following the same pinning approach used for
temporalio/auto-setup. Keep the change scoped to the temporal-ui service so both
Temporal components remain version-controlled and stable across rebuilds.
- Around line 451-465: The worker service repeats the same DATABASE_URL,
POSTGRES_*, and TRINO_* environment values already set on backend, so update
docker-compose to define these shared settings once and reuse them for both
services. Use a YAML anchor/merge pattern around the shared env block in backend
and worker so connection settings stay consistent and only need to be changed in
one place; keep the service-specific variables separate.
- Around line 429-439: The temporal-ui service is forcing insecure CSRF cookies
with TEMPORAL_CSRF_COOKIE_INSECURE=true, which should not be the default in a
production-oriented compose setup. Update the temporal-ui block to make this
setting conditional or configurable via an environment variable so secure
behavior is the default, and keep the insecure option only for explicit local
HTTP usage; use the temporal-ui service definition and its environment entries
to locate the change.
- Around line 448-470: The worker service is inheriting the full backend secret
set through env_file, which gives it more credentials than worker/app/config.py
needs. Update the worker service definition in docker-compose.yml to use a
worker-specific env file instead of ./backend/.env, and scope that file to only
the variables consumed by worker/app/config.py (DB, Temporal, Trino, and LLM
settings). Keep the existing environment entries in the worker service aligned
with the Settings model so no unrelated backend secrets are loaded.
- Around line 414-439: The temporal-ui service is starting before Temporal is
actually ready, since it only waits for service_started on temporal. Add a
readiness healthcheck to the temporal service itself and update temporal-ui’s
depends_on entry to require service_healthy so it only starts after Temporal is
accepting connections; use the existing temporal and temporal-ui service
definitions to make the change.
In `@frontend/src/components/tables/ProfilingTab.tsx`:
- Around line 1340-1352: Add explicit error handling to the ProfilingTab
mutation hooks so failures are surfaced to the user. In the runMutation and
terminateMutation definitions, add onError handlers alongside the existing
onSuccess invalidation, and make them notify/report the failure using the same
UI feedback pattern used elsewhere in ProfilingTab. Use the existing
profilingApi.run, profilingApi.terminate, and qc.invalidateQueries setup as the
anchor points when updating these useMutation calls.
- Around line 1412-1432: The failed/partial action buttons in ProfilingTab are
guarded by an outer isRunning false path, so the inner isRunning checks are
unreachable and misleading. In the ProfilingTab component, remove the redundant
disabled={isRunning} and the spinner/“Running…” ternaries from the
profile?.status === 'failed' || profile?.is_partial block, and replace them with
the static non-running UI state since this branch can never render while
isRunning is true.
In `@worker/app/workflows/profiling_activities.py`:
- Line 53: The profiling workflow is writing naive datetimes in several
persistence points, which can break comparisons with timezone-aware timestamps.
Update the `profiling_activities` code paths that assign `updated_at`,
`cached_until`, and the `computed_at` payload to use
`datetime.now(timezone.utc)` consistently. Use the existing datetime assignments
in the relevant activity methods to standardize all timestamp generation on
UTC-aware values.
- Around line 246-297: The `persist` logic in `profiling_activities.py` deletes
existing `ColumnProfile` rows and then inserts the new set in separate commits,
which can permanently drop data if `_make_json_safe(cs.stats_json)` or a later
insert fails. Move the “Clear old column profiles” delete and the “Persist new
column profiles” insert loop into one database transaction using the existing
`Session`/`session` flow so both succeed or both roll back together. Keep the
`TableProfile` update, delete, and re-insert sequence under the same atomic
unit, and only commit once at the end.
- Around line 71-95: The resumed-column rebuild in profiling_activities.py is
omitting several semantic booleans from existing_columns, so profile JSON is
incomplete after resume. Update the existing_cols_db loop in the
resume_from_partial path to re-derive and include is_large_categorical,
is_free_text, is_boolean, and is_continuous from each
ColumnProfile.semantic_type before appending the saved column payload. Keep the
fix localized to the existing_columns reconstruction used by the profiling
workflow.
In `@worker/app/workflows/profiling_workflow.py`:
- Around line 188-212: The `finally` block in `profiling_workflow` swallows
failures from `persist_profiling_results_activity`, which can leave a
`TableProfile` stuck in `running`. Update the
`persist_profiling_results_activity` call handling so failures are not only
logged: either re-raise the exception after `logger.error(...)` or record the
failure state in workflow-visible metadata so status sync can detect it. Keep
the fix localized around the `finally` persistence path and the
`persist_payload`/`workflow.execute_activity` flow.
- Around line 1-19: Replace the standard logger setup in the workflow module
with Temporal’s replay-safe logger. In the workflow code around
`TableProfilingWorkflow.run`, stop using `logging.getLogger(__name__)` and use
`workflow.logger` instead so logs are suppressed during replay. Keep the
existing workflow/activity structure intact and update any workflow logging
calls to reference the Temporal logger.
In `@worker/Dockerfile`:
- Line 4: The Dockerfile uses an unpinned uv image reference, which makes builds
non-reproducible. Update the COPY --from source image currently using
ghcr.io/astral-sh/uv:latest to a fixed version tag or digest so the build always
pulls the same uv release. Use the existing COPY
--from=ghcr.io/astral-sh/uv:latest instruction as the locator and keep the rest
of the Dockerfile unchanged.
- Around line 1-27: Add a non-root runtime user to the worker image and switch
to it before CMD, since the current Dockerfile leaves the worker process running
as root. Update the Dockerfile around the existing WORKDIR /app/worker, uv sync,
and PATH setup so dependencies can still be installed as root during build, then
create/apply a dedicated USER for the final runtime stage and ensure app
files/venv are accessible to that user.
In `@worker/pyproject.toml`:
- Around line 6-16: The dependency list in pyproject.toml has one inconsistent
version specifier: temporalio is using a lower-bound constraint while the rest
are exact pins. Update the temporalio entry in the dependencies array to use an
exact version pin consistent with the other packages, so rebuilds do not pick up
untested newer releases.
---
Outside diff comments:
In `@backend/app/infra_init.py`:
- Around line 1159-1241: The profile persistence logic in `_run_profile` is
duplicated and has already drifted from
`routers/profiling.py::_run_profile_job_local`, so extract the shared
status/field-copy/`ColumnProfile` save flow into one helper and call it from
both places. Move the common persistence steps (including the `ProfilingStatus`
selection, `is_partial`, profile field assignments, and `ColumnProfile`
creation) into a shared service such as `core.services.profiling_engine`, then
update `_run_profile` to use that helper so both code paths stay consistent.
In `@backend/app/routers/profiling.py`:
- Around line 136-147: The profiling persistence path in the table profile
upsert always marks results as completed, even when the run failed. Update the
status assignment in the Session/TableProfile save block to branch on
result.success so total failures are stored as ProfilingStatus.failed and only
successful runs as completed, while keeping is_partial based on partial
results/errors. Align this logic with the existing conditional status handling
in _run_profile in infra_init.py and use TableProfile.status /
TableProfile.is_partial as the main fields to update.
- Around line 261-283: The profiling endpoints accept a force flag but
run_all_profiles currently drops it when scheduling _trigger_profiling, so the
frontend’s force=true request is ignored. Thread force through the background
task call and into the profiling flow used by
run_table_profile/_trigger_profiling so it can bypass the 24h cache, or remove
the unused parameter from both endpoints and align the API/docs accordingly.
In `@core/src/core/services/profiling_engine.py`:
- Around line 691-753: The `_process_column_metrics` function is too
branch-heavy because it mixes null/distinct fetching with semantic-type-specific
extraction. Refactor it by extracting the shared precomputed-vs-query logic from
`_process_column_metrics`, then replace the long `if/elif` chain with a small
dispatch table or helper handlers for `"categorical"`, `"boolean"`,
`"large_categorical"`, `"continuous"`, and `"time"`. Keep the existing helpers
like `_extract_categorical_stats`, `_extract_continuous_stats`,
`_extract_time_stats`, and the top-values query path, but move each
type-specific block into its own function to simplify the main flow.
In `@worker/app/workflows/profiling_workflow.py`:
- Around line 25-213: The run method in TableProfilingWorkflow is too complex
and should be split into smaller private helpers to satisfy the PLR0912/PLR0915
warning. Extract the chunk-profiling logic, per-column profiling execution, and
AI summary generation into dedicated methods on the same workflow class, then
have run orchestrate those helpers while preserving the existing flow and error
handling.
---
Duplicate comments:
In `@backend/app/routers/profiling.py`:
- Around line 106-111: The profiling flow still silently falls back from
Temporal to local execution, which should be removed; update
`_trigger_profiling` so that if Temporal cannot start, it raises an error
instead of calling `_run_profile_job_local`. Also ensure `resume_from_partial`
is not accepted as a no-op in this fallback path—keep the Temporal-only behavior
explicit in `_trigger_profiling` and `_run_profile_job_local` so users are told
profiling cannot proceed when Temporal is unavailable.
- Around line 88-103: The Temporal workflow start handling in profiling should
stop matching exceptions by stringifying e.g. in the workflow trigger path and
instead catch the actual Temporal exception classes directly. Update the
exception handling around Client.start_workflow in the profiling router to use
explicit exception types such as WorkflowAlreadyStartedError and
WorkflowExecutionAlreadyStartedError, and keep the existing “already running”
branch for those cases. Also reuse a shared Temporal client instead of calling
Client.connect in each of the profiling entry points, including
get_table_profile and terminate_table_profile, ideally by initializing it once
at app startup and injecting it where needed.
🪄 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: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: cc8ec6bc-e69c-4488-a188-3a9c26cc55f2
⛔ Files ignored due to path filters (2)
agent/uv.lockis excluded by!**/*.lockbackend/uv.lockis excluded by!**/*.lock
📒 Files selected for processing (22)
backend/alembic/versions/a123e456b789_add_is_partial_to_table_profiles.pybackend/app/config.pybackend/app/infra_init.pybackend/app/routers/profiling.pybackend/e2e_test.pybackend/pyproject.tomlcore/pyproject.tomlcore/src/core/config.pycore/src/core/models/models.pycore/src/core/services/__init__.pycore/src/core/services/profiling_engine.pydocker-compose.ymlfrontend/src/api/client.tsfrontend/src/components/tables/ProfilingTab.tsxfrontend/src/types/index.tsworker/Dockerfileworker/app/config.pyworker/app/main.pyworker/app/workflows/__init__.pyworker/app/workflows/profiling_activities.pyworker/app/workflows/profiling_workflow.pyworker/pyproject.toml
| # Sync state from Temporal if stuck | ||
| if profile.status in (ProfilingStatus.running, ProfilingStatus.pending): | ||
| try: | ||
| client = await Client.connect(settings.TEMPORAL_HOST) | ||
| handle = client.get_workflow_handle(f"profile-{table_id}") | ||
| desc = await handle.describe() | ||
| if desc.status in ( | ||
| WorkflowExecutionStatus.FAILED, | ||
| WorkflowExecutionStatus.TERMINATED, | ||
| WorkflowExecutionStatus.TIMED_OUT, | ||
| WorkflowExecutionStatus.CANCELED | ||
| ): | ||
| profile.status = ProfilingStatus.failed | ||
| session.add(profile) | ||
| session.commit() | ||
| session.refresh(profile) | ||
| except Exception as e: | ||
| if "NotFound" in str(type(e)): | ||
| profile.status = ProfilingStatus.failed | ||
| session.add(profile) | ||
| session.commit() | ||
| session.refresh(profile) | ||
| else: | ||
| logger.warning("[Profiling] Failed to sync temporal status for %s: %s", table_id, e) | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== profiling.py around relevant lines =="
sed -n '200,280p' backend/app/routers/profiling.py
echo
echo "== Temporal-related imports/usages =="
rg -n "RPCError|RPCStatusCode|NotFound|describe\(\)|get_workflow_handle|Client.connect|WorkflowExecutionStatus" backend -SRepository: StavPonte11/text2sql-onboarding
Length of output: 4502
🌐 Web query:
temporalio Python RPCError NOT_FOUND workflow not found describe get_workflow_handle documentation
💡 Result:
In the Temporal Python SDK, an RPCError with the status NOT_FOUND indicates that the requested Workflow Execution could not be found by the server [1]. Understanding Workflow Handles and NOT_FOUND It is important to note that the method Client.get_workflow_handle does not perform a network call to the Temporal server [2]. It simply creates a local object (a handle) that represents the intent to interact with a specific Workflow ID [3][4]. The NOT_FOUND error is typically raised only when you invoke an operation on that handle that requires communication with the server, such as: - Calling describe to fetch the current status of the Workflow [4]. - Sending a Signal, Query, or Update to the Workflow [1]. - Requesting the result of the Workflow [3][4]. - Attempting to terminate or cancel the Workflow [3]. Common Causes for NOT_FOUND If you receive this error, it generally means the Temporal server does not recognize the specific Workflow Execution: 1. Workflow Does Not Exist: The Workflow ID provided does not correspond to any known, active, or closed workflow in the current namespace [1][2]. 2. Workflow Completed or Expired: The Workflow may have finished or been terminated. If the Workflow has passed its retention period (the time history is kept in the database), it may no longer be available for retrieval via standard API calls [5]. 3. Execution Status Mismatch: In some cases, a workflow might appear in a visibility index but be missing from primary persistence (e.g., due to a data synchronization issue in a self-hosted environment), leading to conflicts where the workflow cannot be described or terminated but also cannot be restarted [6][7]. 4. Namespace Mismatch: Ensure that the Client is connected to the correct namespace where the Workflow was originally started [6]. Handling the Error To handle this exception in your Python code, you can catch the RPCError and inspect the status code [8][1]: from temporalio.client import RPCError, RPCStatusCode try: await handle.describe except RPCError as e: if e.status == RPCStatusCode.NOT_FOUND: print("Workflow not found.") else: raise
Citations:
- 1: https://docs.temporal.io/develop/python/workflows/message-passing
- 2: https://stackoverflow.com/questions/78681428/temporal-skd-does-not-return-all-workflow-executions
- 3: https://python.temporal.io/temporalio.client.WorkflowHandle.html
- 4: https://docs.temporal.io/develop/python/client/temporal-client
- 5: Can't display archived workflows which have past the retention temporalio/temporal#5022
- 6: https://community.temporal.io/t/workflow-execution-not-found-but-cannot-create-new-workflow-of-the-same-id/11584
- 7: https://community.temporal.io/t/workflow-stuck-in-limbo-state/15502
- 8: https://python.temporal.io/temporalio.client.RPCError.html
Handle Temporal NOT_FOUND explicitly
handle.describe() raises an RPCError with status == NOT_FOUND when the workflow is missing; checking str(type(e)) will never match, so stale running/pending profiles stay stuck instead of being marked failed. Catch the RPC error and branch on its status code here.
🤖 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 `@backend/app/routers/profiling.py` around lines 231 - 255, The Temporal sync
block in the profiling router is checking the wrong exception shape, so missing
workflows are not being detected and stuck profiles remain running or pending.
Update the try/except around Client.connect, get_workflow_handle, and
handle.describe to catch the RPCError from Temporal and branch on its status
code for NOT_FOUND, then mark the profile as failed and persist it; keep the
existing failure handling for other workflow terminal states and log unexpected
errors through the existing logger.warning path.
| @router.post("/tables/{table_id}/profile/terminate", status_code=200) | ||
| async def terminate_table_profile(table_id: str, session: Session = Depends(get_session)): | ||
| """ | ||
| Terminates a running Temporal profiling workflow for the given table. | ||
| """ | ||
| table = session.get(Table, table_id) | ||
| profile = session.exec( | ||
| select(TableProfile).where(TableProfile.table_id == table_id) | ||
| ).first() | ||
|
|
||
| if not table or not profile or profile.status not in (ProfilingStatus.running, ProfilingStatus.pending): | ||
| raise HTTPException(status_code=404, detail="Table not found") | ||
|
|
||
| try: | ||
| client = await Client.connect(settings.TEMPORAL_HOST) | ||
| handle = client.get_workflow_handle(f"profile-{table_id}") | ||
| await handle.cancel() | ||
| except Exception as e: | ||
| logger.warning(f"Failed to cancel temporal workflow for {table_id}: {e}") | ||
|
|
||
| profile = session.exec( | ||
| select(TableProfile).where(TableProfile.table_id == table_id) | ||
| ).first() | ||
|
|
||
| if profile: | ||
| profile.status = ProfilingStatus.failed | ||
| session.add(profile) | ||
| session.commit() | ||
|
|
||
| return {"message": "Profiling job terminated", "status": "failed"} | ||
|
|
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== profiling router excerpt ==\n'
sed -n '300,380p' backend/app/routers/profiling.py
printf '\n== search TableProfilingWorkflow and status updates ==\n'
rg -n "TableProfilingWorkflow|is_partial|ProfilingStatus|status =" backend/app -S
printf '\n== search frontend polling/termination logic ==\n'
rg -n "profile/terminate|ProfilingStatus.running|ProfilingStatus.pending|status.*running|status.*pending|poll" frontend backend -SRepository: StavPonte11/text2sql-onboarding
Length of output: 13063
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== profiling workflow/status update path ==\n'
sed -n '120,170p' backend/app/routers/profiling.py
printf '\n== profiling terminate path around line 232 ==\n'
sed -n '220,255p' backend/app/routers/profiling.py
printf '\n== frontend profiling polling logic ==\n'
sed -n '1338,1375p' frontend/src/components/tables/ProfilingTab.tsxRepository: StavPonte11/text2sql-onboarding
Length of output: 4738
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== profile read path ==\n'
sed -n '1,120p' backend/app/routers/profiling.py
printf '\n== frontend profile query source ==\n'
sed -n '1,140p' frontend/src/components/tables/ProfilingTab.tsx
printf '\n== backend profile endpoints ==\n'
rg -n "GET /tables/\{table_id\}/profile|profileQ|invalidateQueries\\(\\{ queryKey: \\['profile'" backend/app frontend/src -SRepository: StavPonte11/text2sql-onboarding
Length of output: 8844
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== infra_init profiling status write ==\n'
sed -n '1148,1210p' backend/app/infra_init.py
printf '\n== workflow/task references ==\n'
rg -n "TableProfilingWorkflow|profile-\\{table_id\\}|cancel\\(|terminated|canceled|failed|completed|is_partial" backend/app/infra_init.py backend/app/routers/profiling.py -SRepository: StavPonte11/text2sql-onboarding
Length of output: 6122
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== workflow definition search ==\n'
rg -n "`@workflow`\.defn|class .*Workflow|TableProfilingWorkflow|workflow\.run|start_workflow\(" backend -S
printf '\n== profile persistence writes ==\n'
rg -n "profile\.status = ProfilingStatus\.(completed|failed)|session\.commit\(\)|run_table_profiling\(" backend/app -SRepository: StavPonte11/text2sql-onboarding
Length of output: 4798
Surface cancellation failures instead of unconditionally marking the profile failed. handle.cancel() can fail, but this endpoint still logs the error, flips profile.status to failed, and returns 200. That leaves the API/UI treating profiling as terminated even when the Temporal workflow may still be running.
🤖 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 `@backend/app/routers/profiling.py` around lines 326 - 356, The
terminate_table_profile endpoint is swallowing cancellation failures and always
marking the TableProfile as failed, even when
Client.connect/get_workflow_handle/handle.cancel does not succeed. Update
terminate_table_profile to surface the exception path from the Temporal cancel
attempt (instead of continuing to success), and only set profile.status to
failed and return the terminated response after a confirmed cancel. Keep the
flow tied to the existing terminate_table_profile, TableProfile, and
handle.cancel logic so the API does not report success when the workflow may
still be running.
| "minio>=7.2.0", | ||
| "core", | ||
| "mcp>=1.2.0", | ||
| "temporalio>=1.6.0", |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Consider pinning temporalio to an exact/bounded version.
temporalio>=1.6.0 has no upper bound; an unvetted future major/minor bump could break the worker without warning. Other deps in this repo (e.g., sqlmodel==0.0.22) are pinned exactly.
🤖 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 `@backend/pyproject.toml` at line 28, The backend dependency list currently
allows temporalio to float with only a lower bound, which risks untested
upgrades breaking the worker. Update the dependency entry in pyproject.toml for
temporalio to use an exact pin or a bounded version range consistent with other
locked dependencies like sqlmodel, so future installs remain predictable.
There was a problem hiding this comment.
yes he's right, change to fixed version
There was a problem hiding this comment.
🧩 Analysis chain
🤖 web_search executed:
What is the latest release version of the temporalio Python SDK on PyPI?
Length of output: 2085
@yuvalkh, sounds good! For reference, the latest stable release is temporalio==1.30.0 on PyPI. You could pin it like this:
"temporalio==1.30.0",Alternatively, if you'd like some flexibility for patch-level updates while still avoiding untested major/minor bumps, a bounded range like temporalio>=1.6.0,<2.0.0 would also work, consistent with how tightly other deps in this file are pinned (e.g., sqlmodel==0.0.22).
ᕦ(ò_óˇ)ᕤ
| "langchain", | ||
| "langchain-openai", |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Pin langchain/langchain-openai versions.
These are added with no version constraint whatsoever, risking unreproducible builds and unannounced breaking changes (both packages iterate quickly). Pin to a known-compatible range, consistent with other pinned deps like sqlmodel==0.0.22.
🤖 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 `@core/pyproject.toml` around lines 15 - 16, The dependency entries for
langchain and langchain-openai are unpinned, which can lead to non-reproducible
builds and unexpected breakage. Update the dependency definitions in
pyproject.toml to use a known-compatible version range or exact pins, matching
the project’s existing pinned dependency style like sqlmodel==0.0.22, so the
package resolver always installs the intended versions.
| if any(t in dtype_lower for t in NUMERIC_TYPES): | ||
| select_parts.append(f'MIN({safe_col}) AS "{col_name}__min"') | ||
| select_parts.append(f'MAX({safe_col}) AS "{col_name}__max"') | ||
| select_parts.append(f'AVG(TRY_CAST({safe_col} AS DOUBLE)) AS "{col_name}__avg"') | ||
| select_parts.append(f'APPROX_PERCENTILE(TRY_CAST({safe_col} AS DOUBLE), ARRAY[0.25, 0.5, 0.75]) AS "{col_name}__quants"') | ||
| select_parts.append(f'STDDEV_POP(TRY_CAST({safe_col} AS DOUBLE)) AS "{col_name}__std"') | ||
| elif any(t in dtype_lower for t in TIME_TYPES): | ||
| select_parts.append(f'MIN({safe_col}) AS "{col_name}__min"') | ||
| select_parts.append(f'MAX({safe_col}) AS "{col_name}__max"') | ||
| select_parts.append(f'APPROX_PERCENTILE(to_unixtime(TRY_CAST({safe_col} AS TIMESTAMP)), ARRAY[0.25, 0.5, 0.75]) AS "{col_name}__quants"') | ||
| select_parts.append(f'STDDEV_POP(to_unixtime(TRY_CAST({safe_col} AS TIMESTAMP))) AS "{col_name}__std"') | ||
| select_parts.append(f'MIN(to_unixtime(TRY_CAST({safe_col} AS TIMESTAMP))) AS "{col_name}__min_unix"') | ||
| select_parts.append(f'MAX(to_unixtime(TRY_CAST({safe_col} AS TIMESTAMP))) AS "{col_name}__max_unix"') |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -n 'TIME_TYPES\s*=' core/src/core/services/profiling_engine.pyRepository: StavPonte11/text2sql-onboarding
Length of output: 187
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file=core/src/core/services/profiling_engine.py
echo '--- TIME_TYPES and related constants ---'
sed -n '70,120p' "$file"
echo
echo '--- _classify_semantic_type ---'
rg -n 'def _classify_semantic_type|is_time =|TIME_HINTS|NUMERIC_TYPES|TIME_TYPES|build_combined_profiling_query' "$file" -n -A40 -B10
echo
echo '--- classification call sites ---'
rg -n '_classify_semantic_type\(' -A3 -B3 "$file"Repository: StavPonte11/text2sql-onboarding
Length of output: 15830
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file=core/src/core/services/profiling_engine.py
echo '--- _analyze_column around semantic_type usage ---'
sed -n '650,790p' "$file"
echo
echo '--- parse_combined_result time branch ---'
sed -n '336,375p' "$file"Repository: StavPonte11/text2sql-onboarding
Length of output: 7995
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file=core/src/core/services/profiling_engine.py
echo '--- build_column_metadata_query ---'
sed -n '220,290p' "$file"
echo
echo '--- any type normalization before classification ---'
rg -n 'lower\(\)|data_type|field_type|field_name|build_column_metadata_query|information_schema' "$file" -A3 -B3Repository: StavPonte11/text2sql-onboarding
Length of output: 14784
Align _classify_semantic_type with the combined-query time check Parameterized Trino types like timestamp(3) match the combined-query time branch, but this exact-equality check misses them, so they can be classified as non-time and skip time-specific stats. Use the same substring match here.
🤖 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 `@core/src/core/services/profiling_engine.py` around lines 316 - 328, The
time-type handling in profiling_engine’s combined-query stats block is using an
exact match that misses parameterized Trino timestamp types like timestamp(3).
Update the time branch in the profiling logic that builds select_parts to use
the same substring-based check as _classify_semantic_type, so all TIME_TYPES
variants are recognized and time-specific aggregates are generated consistently.
| import asyncio | ||
| import logging | ||
| from datetime import timedelta | ||
|
|
||
| from temporalio import workflow | ||
| from temporalio.common import RetryPolicy | ||
|
|
||
| # Import activity definitions and config bypassing sandbox restrictions | ||
| with workflow.unsafe.imports_passed_through(): | ||
| from app.config import settings | ||
| from app.workflows.profiling_activities import ( | ||
| compute_chunk_metrics_activity, | ||
| fetch_table_metadata_activity, | ||
| generate_ai_summary_activity, | ||
| persist_profiling_results_activity, | ||
| profile_column_activity, | ||
| ) | ||
|
|
||
| logger = logging.getLogger(__name__) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Use workflow.logger instead of the standard logging module inside workflow code.
Temporal's Python SDK docs state The SDK provides replay-safe alternatives for common needs: Use workflow.logger instead of print() or the standard logging module. The SDK logger automatically suppresses log messages during replay to avoid duplicates. Using logging.getLogger(__name__) inside TableProfilingWorkflow.run will re-emit every log line on each history replay, producing noisy/duplicated logs in production.
🤖 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 `@worker/app/workflows/profiling_workflow.py` around lines 1 - 19, Replace the
standard logger setup in the workflow module with Temporal’s replay-safe logger.
In the workflow code around `TableProfilingWorkflow.run`, stop using
`logging.getLogger(__name__)` and use `workflow.logger` instead so logs are
suppressed during replay. Keep the existing workflow/activity structure intact
and update any workflow logging calls to reference the Temporal logger.
| finally: | ||
| # 5. Persist results activity | ||
| persist_payload = { | ||
| "table_id": table_id, | ||
| "profile_id": profile_id, | ||
| "table_fqn": fqn, | ||
| "row_count": row_count, | ||
| "sample_size": sample_size, | ||
| "column_count": len(columns_meta), | ||
| "sample_data": sample_data, | ||
| "column_stats": column_stats, | ||
| "ai_summary": ai_summary, | ||
| "is_partial": is_partial, | ||
| "failed_subtasks": failed_subtasks, | ||
| "errors": errors, | ||
| } | ||
| try: | ||
| await workflow.execute_activity( | ||
| persist_profiling_results_activity, | ||
| persist_payload, | ||
| start_to_close_timeout=timedelta(seconds=settings.ACTIVITY_START_TO_CLOSE_TIMEOUT_SECONDS), | ||
| retry_policy=RetryPolicy(maximum_attempts=3), | ||
| ) | ||
| except Exception as p_exc: | ||
| logger.error("Failed to persist results in finally block: %s", p_exc) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Persist failures in the finally block are swallowed, leaving the profile stuck at running forever.
fetch_table_metadata_activity sets profile.status = ProfilingStatus.running early on; only persist_profiling_results_activity ever advances it to completed. If persistence fails after all 3 retries (e.g. due to the non-atomic delete/insert issue noted in profiling_activities.py), the exception here is only logged (logger.error(...)) and not re-raised or otherwise surfaced — the workflow function then returns normally. The TableProfile row is permanently left in running status with no automatic recovery path, and the workflow itself reports success.
Consider re-raising (so the workflow is marked failed and can be retried/resumed) or recording the failure via a workflow query/search-attribute so the backend's status-sync logic can detect and surface it.
🧰 Tools
🪛 Ruff (0.15.20)
[warning] 211-211: Do not catch blind exception: Exception
(BLE001)
[warning] 212-212: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
🤖 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 `@worker/app/workflows/profiling_workflow.py` around lines 188 - 212, The
`finally` block in `profiling_workflow` swallows failures from
`persist_profiling_results_activity`, which can leave a `TableProfile` stuck in
`running`. Update the `persist_profiling_results_activity` call handling so
failures are not only logged: either re-raise the exception after
`logger.error(...)` or record the failure state in workflow-visible metadata so
status sync can detect it. Keep the fix localized around the `finally`
persistence path and the `persist_payload`/`workflow.execute_activity` flow.
| FROM python:3.12-slim | ||
|
|
||
| # Install uv, git and openssh-client | ||
| COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv | ||
| RUN apt-get update && apt-get install -y --no-install-recommends git openssh-client && rm -rf /var/lib/apt/lists/* | ||
|
|
||
| # Configure git to use ssh for github.com | ||
| RUN git config --global url."git@github.com:".insteadOf "https://github.com/" | ||
|
|
||
| WORKDIR /app | ||
|
|
||
| # Copy workspace dependencies | ||
| COPY core /app/core | ||
| COPY worker /app/worker | ||
|
|
||
| WORKDIR /app/worker | ||
|
|
||
| # Mount SSH deploy key secret and sync dependencies | ||
| RUN --mount=type=secret,id=deploy_key,target=/root/.ssh/id_rsa,mode=0600 \ | ||
| mkdir -p -m 0700 /root/.ssh && \ | ||
| ssh-keyscan github.com >> /root/.ssh/known_hosts && \ | ||
| uv sync --no-dev --no-install-project | ||
|
|
||
| # Add virtualenv bin to PATH | ||
| ENV PATH="/app/worker/.venv/bin:$PATH" | ||
|
|
||
| CMD ["python", "-m", "app.main"] |
There was a problem hiding this comment.
🔒 Security & Privacy | 🔵 Trivial | ⚡ Quick win
Container runs as root.
No USER directive is set, so the worker process (and any dependency vulnerability) runs with root privileges inside the container. Static analysis (Trivy DS-0002, Checkov CKV_DOCKER_3) both flag this.
🔒 Proposed fix
ENV PATH="/app/worker/.venv/bin:$PATH"
+RUN useradd --create-home --shell /bin/bash worker && \
+ chown -R worker:worker /app
+USER worker
+
CMD ["python", "-m", "app.main"]📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| FROM python:3.12-slim | |
| # Install uv, git and openssh-client | |
| COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv | |
| RUN apt-get update && apt-get install -y --no-install-recommends git openssh-client && rm -rf /var/lib/apt/lists/* | |
| # Configure git to use ssh for github.com | |
| RUN git config --global url."git@github.com:".insteadOf "https://github.com/" | |
| WORKDIR /app | |
| # Copy workspace dependencies | |
| COPY core /app/core | |
| COPY worker /app/worker | |
| WORKDIR /app/worker | |
| # Mount SSH deploy key secret and sync dependencies | |
| RUN --mount=type=secret,id=deploy_key,target=/root/.ssh/id_rsa,mode=0600 \ | |
| mkdir -p -m 0700 /root/.ssh && \ | |
| ssh-keyscan github.com >> /root/.ssh/known_hosts && \ | |
| uv sync --no-dev --no-install-project | |
| # Add virtualenv bin to PATH | |
| ENV PATH="/app/worker/.venv/bin:$PATH" | |
| CMD ["python", "-m", "app.main"] | |
| FROM python:3.12-slim | |
| # Install uv, git and openssh-client | |
| COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv | |
| RUN apt-get update && apt-get install -y --no-install-recommends git openssh-client && rm -rf /var/lib/apt/lists/* | |
| # Configure git to use ssh for github.com | |
| RUN git config --global url."git@github.com:".insteadOf "https://github.com/" | |
| WORKDIR /app | |
| # Copy workspace dependencies | |
| COPY core /app/core | |
| COPY worker /app/worker | |
| WORKDIR /app/worker | |
| # Mount SSH deploy key secret and sync dependencies | |
| RUN --mount=type=secret,id=deploy_key,target=/root/.ssh/id_rsa,mode=0600 \ | |
| mkdir -p -m 0700 /root/.ssh && \ | |
| ssh-keyscan github.com >> /root/.ssh/known_hosts && \ | |
| uv sync --no-dev --no-install-project | |
| # Add virtualenv bin to PATH | |
| ENV PATH="/app/worker/.venv/bin:$PATH" | |
| RUN useradd --create-home --shell /bin/bash worker && \ | |
| chown -R worker:worker /app | |
| USER worker | |
| CMD ["python", "-m", "app.main"] |
🧰 Tools
🪛 Checkov (3.3.2)
[low] 1-27: Ensure that HEALTHCHECK instructions have been added to container images
(CKV_DOCKER_2)
[low] 1-27: Ensure that a user for the container has been created
(CKV_DOCKER_3)
🪛 Hadolint (2.14.0)
[warning] 5-5: Pin versions in apt get install. Instead of apt-get install <package> use apt-get install <package>=<version>
(DL3008)
[warning] 19-19: When used with -p, -m only applies to the deepest directory.
(SC2174)
🪛 Trivy (0.69.3)
[error] 1-1: Image user should not be 'root'
Specify at least 1 USER command in Dockerfile with non-root user as argument
Rule: DS-0002
(IaC/Dockerfile)
[info] 1-1: No HEALTHCHECK defined
Add HEALTHCHECK instruction in your Dockerfile
Rule: DS-0026
(IaC/Dockerfile)
🤖 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 `@worker/Dockerfile` around lines 1 - 27, Add a non-root runtime user to the
worker image and switch to it before CMD, since the current Dockerfile leaves
the worker process running as root. Update the Dockerfile around the existing
WORKDIR /app/worker, uv sync, and PATH setup so dependencies can still be
installed as root during build, then create/apply a dedicated USER for the final
runtime stage and ensure app files/venv are accessible to that user.
Source: Linters/SAST tools
| FROM python:3.12-slim | ||
|
|
||
| # Install uv, git and openssh-client | ||
| COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Pin the uv base image tag.
ghcr.io/astral-sh/uv:latest is unpinned, so builds are not reproducible and can silently break when a new uv release changes 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 `@worker/Dockerfile` at line 4, The Dockerfile uses an unpinned uv image
reference, which makes builds non-reproducible. Update the COPY --from source
image currently using ghcr.io/astral-sh/uv:latest to a fixed version tag or
digest so the build always pulls the same uv release. Use the existing COPY
--from=ghcr.io/astral-sh/uv:latest instruction as the locator and keep the rest
of the Dockerfile unchanged.
| dependencies = [ | ||
| "temporalio>=1.6.0", | ||
| "pydantic==2.10.4", | ||
| "pydantic-settings==2.7.0", | ||
| "python-dotenv==1.0.1", | ||
| "sqlmodel==0.0.22", | ||
| "trino==0.328.0", | ||
| "psycopg2-binary==2.9.9", | ||
| "pandas==2.2.3", | ||
| "core", | ||
| ] |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Inconsistent dependency pinning.
All other dependencies use exact pins (==), but temporalio uses >=1.6.0, allowing an untested newer major/minor version to be installed on rebuild.
Proposed fix
- "temporalio>=1.6.0",
+ "temporalio==1.6.0",📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| dependencies = [ | |
| "temporalio>=1.6.0", | |
| "pydantic==2.10.4", | |
| "pydantic-settings==2.7.0", | |
| "python-dotenv==1.0.1", | |
| "sqlmodel==0.0.22", | |
| "trino==0.328.0", | |
| "psycopg2-binary==2.9.9", | |
| "pandas==2.2.3", | |
| "core", | |
| ] | |
| dependencies = [ | |
| "temporalio==1.6.0", | |
| "pydantic==2.10.4", | |
| "pydantic-settings==2.7.0", | |
| "python-dotenv==1.0.1", | |
| "sqlmodel==0.0.22", | |
| "trino==0.328.0", | |
| "psycopg2-binary==2.9.9", | |
| "pandas==2.2.3", | |
| "core", | |
| ] |
🤖 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 `@worker/pyproject.toml` around lines 6 - 16, The dependency list in
pyproject.toml has one inconsistent version specifier: temporalio is using a
lower-bound constraint while the rest are exact pins. Update the temporalio
entry in the dependencies array to use an exact version pin consistent with the
other packages, so rebuilds do not pick up untested newer releases.
| select(TableProfile).where(TableProfile.table_id == table_id) | ||
| ).first() | ||
|
|
||
| if not table or not profile or profile.status not in (ProfilingStatus.running, ProfilingStatus.pending): |
There was a problem hiding this comment.
I'm not sure but we don't change to running or pending no? Because if we already have a completed profile, when we run again it won't be running or pending because we want to save the previous one (only save the latest in DB)
|
|
||
|
|
||
| @activity.defn | ||
| def fetch_table_metadata_activity(payload: dict) -> dict: |
There was a problem hiding this comment.
Change from payload into 2 parameters: table_id and resume_from_partial
|
|
||
|
|
||
| @activity.defn | ||
| def persist_profiling_results_activity(payload: dict) -> None: |
There was a problem hiding this comment.
In this function there are some magic numbers, so maybe change them to constants at start
| }) | ||
|
|
||
| # 4. Generate AI summary activity (skip if no columns completed or it's a completely failed run) | ||
| ai_summary = "" |
There was a problem hiding this comment.
Maybe you meant like auto_insights that we had before, but now I'm not sure we need it because we just want raw metadata about the columns and not something smarter
3 participants
Summary by CodeRabbit
New Features
Bug Fixes
Chores