Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions functions-python/batch_process_dataset/src/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@
from cloudevents.http import CloudEvent
from google.cloud import storage
from sqlalchemy import func

from shared.common.gcp_utils import create_refresh_materialized_view_task
from shared.database_gen.sqlacodegen_models import Gtfsdataset, Gtfsfile

Expand All @@ -37,9 +36,10 @@
import logging

from shared.helpers.logger import init_logger, get_logger
from shared.helpers.utils import download_and_get_hash
from shared.helpers.utils import download_and_get_hash, get_hash_from_file
from sqlalchemy.orm import Session


init_logger()


Expand Down Expand Up @@ -179,6 +179,8 @@ def upload_file_to_storage(
id=str(uuid.uuid4()),
file_name=file_name,
file_size_bytes=os.path.getsize(file_path),
hosted_url=file_blob.public_url if public else None,
hash=get_hash_from_file(file_path),
)
)
return blob, extracted_files
Expand Down
17 changes: 17 additions & 0 deletions functions-python/helpers/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import hashlib
import logging
import os
import ssl

import requests
import urllib3
Expand Down Expand Up @@ -78,6 +79,17 @@ def download_url_content(url, with_retry=False):
raise e


def get_hash_from_file(file_path, hash_algorithm="sha256", chunk_size=8192):
"""
Returns the hash of a file
"""
hash_object = hashlib.new(hash_algorithm)
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(chunk_size), b""):
hash_object.update(chunk)
return hash_object.hexdigest()


def download_and_get_hash(
url,
file_path,
Expand All @@ -87,6 +99,7 @@ def download_and_get_hash(
api_key_parameter_name=None,
credentials=None,
logger=None,
trusted_certs=False, # If True, disables SSL verification
):
"""
Downloads the content of a URL and stores it in a file and returns the hash of the file
Expand Down Expand Up @@ -117,6 +130,10 @@ def download_and_get_hash(
if authentication_type == 2 and api_key_parameter_name and credentials:
headers[api_key_parameter_name] = credentials

if trusted_certs:
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE

with urllib3.PoolManager(ssl_context=ctx) as http:
with http.request(
"GET", url, preload_content=False, headers=headers, redirect=True
Expand Down
7 changes: 3 additions & 4 deletions functions-python/tasks_executor/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@ Example:
"filter_statuses": ["active", "inactive", "future"]
}
}
```
```json
Comment thread
cka-y marked this conversation as resolved.
{
"task": "rebuild_missing_bounding_boxes",
"payload": {
Expand All @@ -40,12 +42,9 @@ Example:
```

To get the list of supported tasks use:
``
```json
{
"name": "list_tasks",
"payload": {}
}

```

```
4 changes: 4 additions & 0 deletions functions-python/tasks_executor/function_config.json
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,10 @@
"secret_environment_variables": [
{
"key": "FEEDS_DATABASE_URL"
},
{
"key": "FEEDS_CREDENTIALS",

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💪

"secret": "FEEDS_CREDENTIALS"
}
],
"ingress_settings": "ALLOW_ALL",
Expand Down
1 change: 1 addition & 0 deletions functions-python/tasks_executor/requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ google-cloud-workflows
google-cloud-pubsub
google-cloud-tasks
flask
google-cloud-storage

# Configuration
python-dotenv==1.0.0
7 changes: 7 additions & 0 deletions functions-python/tasks_executor/src/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@
from tasks.refresh_feedsearch_view.refresh_materialized_view import (
refresh_materialized_view_handler,
)
from tasks.dataset_files.rebuild_missing_dataset_files import (
rebuild_missing_dataset_files_handler,
)
from tasks.validation_reports.rebuild_missing_validation_reports import (
rebuild_missing_validation_reports_handler,
)
Expand Down Expand Up @@ -56,6 +59,10 @@
"description": "Refreshes the materialized view.",
"handler": refresh_materialized_view_handler,
},
"rebuild_missing_dataset_files": {
"description": "Rebuilds missing dataset files for GTFS datasets.",
"handler": rebuild_missing_dataset_files_handler,
},
}


Expand Down
72 changes: 72 additions & 0 deletions functions-python/tasks_executor/src/tasks/dataset_files/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
# Rebuild Missing Dataset Files

This task rebuilds missing extracted files in GTFS datasets.
It downloads datasets from their `hosted_url`, extracts all files, computes zipped and unzipped sizes, calculates hashes, uploads the files to a GCS bucket, and updates the database.

---

## Task ID

Use task ID: `rebuild_missing_dataset_files`

---

## Usage

The function accepts the following payload:

```json
{
"dry_run": true, // [optional] If true, do not upload or modify the database (default: true)
"after_date": "YYYY-MM-DD", // [optional] Only include datasets downloaded after this ISO date
"latest_only": true // [optional] If true, only process the latest version of each dataset (default: true)
Comment thread
cka-y marked this conversation as resolved.
}
```

### Example:

```json
{
"dry_run": false,
"after_date": "2025-07-01",
"latest_only": true
}
```

---

## What It Does

For each GTFS dataset with missing file information (missing zipped/unzipped sizes or missing extracted files), this function:

1. Downloads the `.zip` file from its `hosted_url`
2. Computes the zipped size in bytes
3. Extracts all GTFS files locally
4. Computes the unzipped size in bytes
5. Uploads each extracted file to a GCS bucket with the structure:

```
{feed-stable-id}/{dataset-stable-id}/extracted/{file_name}
```
6. Makes each file publicly accessible and stores its GCS URL
7. Computes SHA256 hashes for each file
8. Stores metadata in the `Gtfsfile` table for later use

---

## GCP Environment Variables

The function requires the following environment variables:

| Variable | Description |
| ---------------------- | ---------------------------------------------------------------------------- |
| `DATASETS_BUCKET_NAME` | The name of the GCS bucket used to store extracted GTFS files |

---

## Additional Notes

* This function **disables SSL verification** when downloading files, as the sources are trusted internally.
* Commits to the database occur in batches of 5 datasets to improve performance and avoid large transaction blocks.
* If `dry_run` is enabled, no downloads, uploads, or DB modifications are performed. Only the number of affected datasets is logged.
* The function is safe to rerun. It will only affect datasets missing required file metadata.
Loading