feat: implement validate API key check endpoint and GCS verification#616
feat: implement validate API key check endpoint and GCS verification#616yiyche wants to merge 2 commits into
Conversation
Not up to standards ⛔🔴 Issues
|
| Category | Results |
|---|---|
| ErrorProne | 1 high |
| CodeStyle | 1 minor |
🟢 Metrics 62 complexity
Metric Results Complexity 62
NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new ingestion router with /start and /config endpoints to validate, trigger, and inspect Cloud Run ingestion jobs. It also adds GCS bucket and file existence checks to the storage client, updates dependencies, and includes comprehensive unit tests. The review feedback focuses on improving robustness and error handling: guarding against potential AttributeErrors when auto-discovering job configurations, logging warnings for failed Cloud Run API requests, and updating error messages to account for permission issues during GCS bucket and file checks.
| if job_resp.status_code == 200: | ||
| job_data = job_resp.json() | ||
| containers = job_data.get("template", {}).get("template", {}).get("containers", []) | ||
| if containers: | ||
| env_vars = containers[0].get("env", []) | ||
| for env_var in env_vars: | ||
| if env_var.get("name") == "GCS_INPUT_FOLDER": | ||
| input_dir = env_var.get("value") | ||
| break | ||
| elif env_var.get("name") == "INPUT_DIR": | ||
| parts = env_var.get("value").removeprefix("gs://").split("/", 1) | ||
| if len(parts) > 1: | ||
| input_dir = parts[1] | ||
| break |
There was a problem hiding this comment.
When auto-discovering the input directory from the Cloud Run job, env_var.get('value') can be None if the environment variable is defined using valueFrom (e.g., referencing a secret or configmap). Guard against None values before calling .removeprefix() to prevent an AttributeError. Additionally, if the API call to fetch job details returns a non-200 status code, logging a warning with the status code and response text will make debugging configuration or permission issues much easier.
| if job_resp.status_code == 200: | |
| job_data = job_resp.json() | |
| containers = job_data.get("template", {}).get("template", {}).get("containers", []) | |
| if containers: | |
| env_vars = containers[0].get("env", []) | |
| for env_var in env_vars: | |
| if env_var.get("name") == "GCS_INPUT_FOLDER": | |
| input_dir = env_var.get("value") | |
| break | |
| elif env_var.get("name") == "INPUT_DIR": | |
| parts = env_var.get("value").removeprefix("gs://").split("/", 1) | |
| if len(parts) > 1: | |
| input_dir = parts[1] | |
| break | |
| if job_resp.status_code == 200: | |
| job_data = job_resp.json() | |
| containers = job_data.get("template", {}).get("template", {}).get("containers", []) | |
| if containers: | |
| env_vars = containers[0].get("env", []) | |
| for env_var in env_vars: | |
| if env_var.get("name") == "GCS_INPUT_FOLDER": | |
| input_dir = env_var.get("value") | |
| break | |
| elif env_var.get("name") == "INPUT_DIR": | |
| val = env_var.get("value") | |
| if val: | |
| parts = val.removeprefix("gs://").split("/", 1) | |
| if len(parts) > 1: | |
| input_dir = parts[1] | |
| break | |
| else: | |
| logging.warning(f"Failed to fetch Cloud Run job details for auto-discovery (HTTP {job_resp.status_code}): {job_resp.text}") |
| if exec_resp.status_code == 200: | ||
| executions_data = exec_resp.json() | ||
| for execution in executions_data.get("executions", []): | ||
| # If completionTime is missing, the execution is actively running or pending | ||
| if "completionTime" not in execution: | ||
| raise HTTPException( | ||
| status_code=409, | ||
| detail=f"An ingestion job execution is already running: {execution.get('name')}" | ||
| ) |
There was a problem hiding this comment.
If the API call to query active executions returns a non-200 status code (e.g., due to permission or configuration issues), it currently fails silently. Logging a warning with the status code and response text will help operators diagnose why the execution check failed.
| if exec_resp.status_code == 200: | |
| executions_data = exec_resp.json() | |
| for execution in executions_data.get("executions", []): | |
| # If completionTime is missing, the execution is actively running or pending | |
| if "completionTime" not in execution: | |
| raise HTTPException( | |
| status_code=409, | |
| detail=f"An ingestion job execution is already running: {execution.get('name')}" | |
| ) | |
| if exec_resp.status_code == 200: | |
| executions_data = exec_resp.json() | |
| for execution in executions_data.get("executions", []): | |
| # If completionTime is missing, the execution is actively running or pending | |
| if "completionTime" not in execution: | |
| raise HTTPException( | |
| status_code=409, | |
| detail=f"An ingestion job execution is already running: {execution.get('name')}" | |
| ) | |
| else: | |
| logging.warning(f"Failed to query active executions (HTTP {exec_resp.status_code}): {exec_resp.text}") |
| if not storage.check_bucket_exists(bucket_name): | ||
| raise HTTPException( | ||
| status_code=400, | ||
| detail=f"GCS bucket '{bucket_name}' does not exist." | ||
| ) |
There was a problem hiding this comment.
If the GCS bucket exists but the service account lacks permission to access it, check_bucket_exists will catch the exception and return False. Raising an HTTPException stating that the bucket 'does not exist' is misleading in this case. Update the error message to indicate that the bucket may either not exist or not be accessible.
| if not storage.check_bucket_exists(bucket_name): | |
| raise HTTPException( | |
| status_code=400, | |
| detail=f"GCS bucket '{bucket_name}' does not exist." | |
| ) | |
| if not storage.check_bucket_exists(bucket_name): | |
| raise HTTPException( | |
| status_code=400, | |
| detail=f"GCS bucket '{bucket_name}' does not exist or is not accessible." | |
| ) |
yiyche
force-pushed
the
feature/validate-api-key
branch
3 times, most recently
from
fb6385d to
bdacd19
Compare
yiyche
force-pushed
the
feature/validate-api-key
branch
from
911ebb5 to
b774d21
Compare
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
| if not storage.check_bucket_exists(bucket_name): | ||
| raise HTTPException( | ||
| status_code=400, | ||
| detail=f"GCS bucket '{bucket_name}' does not exist." | ||
| ) |
| detail=f"Failed to load GCP credentials: {e}" | ||
| ) | ||
|
|
||
| _check_active_executions(session, full_job_name) |
There was a problem hiding this comment.
Can you skip this step for now? It may make sense to add, but for now i want to keep the same behavior as when we start the job via the CLI today.
| """Resolves the full Cloud Run job name resource.""" | ||
| job_name = job_name_in or os.environ.get("INGESTION_JOB_NAME") | ||
| if not job_name: | ||
| raise HTTPException( |
There was a problem hiding this comment.
For a follow-up: we should be packaging these exceptions and returning them as json with the specified status code and messages so they can be easily parsed.
| return job_name | ||
|
|
||
|
|
||
| def _auto_discover_input_dir(full_job_name: str) -> Optional[str]: |
There was a problem hiding this comment.
Instead of auto-resolving the input dir, can we just pass nothing to the cloud run job and have it use its own environment defaults?
| ): | ||
| """Validates parameters, checks pre-conditions, and triggers the Cloud Run job.""" | ||
| full_job_name = _resolve_job_name(req.job_name) | ||
| bucket_name, input_dir = _resolve_gcs_input(req, storage, full_job_name) |
There was a problem hiding this comment.
I talked with @gmechali about this, and Let's move the config.json file existence check out of _resolve_gcs_input and delegate it to the preprocessing job itself, as it will perform this validation at startup anyway. Keeping this out of the helper service keeps this change small and focused purely on API key validation and triggering the job execution.
2 participants
mainly added