Deployments and scheduling

migra/compare.py

@@ -8,17 +8,25 @@
 target_url = settings.DATABASE_WRITE_URL
 
 today = date.today().strftime('%Y%m%d')
-a = db(production_url)
-b = db(target_url)
+b = db(production_url)
+a = db(target_url)
 
-public = b.schemadiff_as_statements(a, schema='public')
-logs = b.schemadiff_as_statements(a, schema='logs')
-deployments = b.schemadiff_as_statements(a, schema='deployments')
+schemas = [
+    'public',
+    'logs',
+    'fetcher'
+    ]
 
+
+## get diff from db
+diff = {}
+for schema in schemas:
+    diff[schema] = b.schemadiff_as_statements(a, schema=schema)
+
+## write out the statements
 with open(f"diff_{today}.sql", "w") as f:
-    for statement in logs:
-        f.write(f"----------------\n{statement}\n")
-    for statement in public:
-        f.write(f"----------------\n{statement}\n")
-    for statement in deployments:
-        f.write(f"----------------\n{statement}\n")
+    for (schema, statements) in diff.items():
+        print(f"printing {schema} - {len(statements)} statements")
+        f.write(f"----------------\n{schema}\n")
+        for statement in statements:
+            f.write(f"----------------\n{statement}\n")

openaqdb/fetcher/README.md

@@ -0,0 +1,93 @@
+# Fetcher Schema
+
+Database-driven scheduling and orchestration system for OpenAQ data adapter deployments.
+
+## Purpose
+
+The fetcher schema manages scheduling and configuration for data adapter deployments. Schedule definitions, adapter assignments, and execution metadata live in the database, enabling runtime configuration changes without application redeployment.
+
+## Design Goals
+
+1. **Database-driven configuration**: All scheduling and deployment settings stored in database tables
+2. **Flexible scheduling**: Cron expressions provide fine-grained control over execution timing
+3. **Observability**: Track deployment history, adapter assignments, and execution metadata
+4. **Queue-based execution**: fetchlogs serves as job queue for polling-based adapter execution
+
+## Architecture
+
+### Core Tables
+
+- **handlers**: SQS queue definitions for routing deployment execution
+- **adapter_clients**: Available adapter implementations (air4thai, clarity, etc.)
+- **adapters**: Links providers to adapter clients with provider-specific configuration
+- **deployments**: Scheduling definitions with cron expressions and temporal offsets
+- **deployment_adapters**: Many-to-many junction linking deployments to adapters
+
+### Execution Flow
+
+1. **pg_cron** runs `queue_deployments()` every minute
+2. Function evaluates cron schedules for active deployments
+3. Ready deployments (with adapters assigned) are inserted into `public.fetchlogs` as scheduled jobs
+4. **Adapter application** polls fetchlogs for new jobs
+5. Application executes adapters, creates file, uploads to S3
+6. **S3 trigger** invokes Lambda to upsert file metadata to fetchlogs
+7. **Ingest application** processes files and updates fetchlogs metadata
+
+### Key Design Choices
+
+**Cron expressions**: Standard 5-field format (minute hour day month weekday) with custom validation functions to ensure correctness before storage.
+
+**fetchlogs as queue**: The `public.fetchlogs` table serves as both execution queue and audit log. The `scheduled_datetime` field marks queued jobs; `loaded_datetime` and `completed_datetime` track progress through the pipeline.
+
+**Unique keys for idempotency**: Fetchlog keys use format `YYYY-MM-DD/prefix/prefix-YYYYMMDDHH24MI`. Each deployment+time combination produces exactly one job via ON CONFLICT DO NOTHING constraint.
+
+**Temporal offsets**: The `temporal_offset` field (in hours) tells adapters to fetch data from N hours in the past, accommodating data sources that publish with delay.
+
+**Two-function design**: `get_ready_deployments()` queries which deployments should run (read-only); `queue_deployments()` inserts them into fetchlogs. The separation enables inspection without side effects.
+
+**Adapter filtering**: Deployments without assigned adapters appear in `get_ready_deployments()` for visibility but are not queued by `queue_deployments()`.
+
+## Usage
+
+### Query ready deployments (inspection/testing)
+```sql
+-- See what's ready to run now
+SELECT * FROM fetcher.get_ready_deployments();
+
+-- Check what would run at specific time
+SELECT * FROM fetcher.get_ready_deployments('2026-01-27 14:30:00');
+```
+
+### Queue deployments (production)
+```sql
+-- Manually queue (typically called by pg_cron)
+SELECT fetcher.queue_deployments();
+```
+
+### Add a new deployment
+```sql
+-- Create deployment running every 15 minutes
+INSERT INTO fetcher.deployments (
+  handlers_id, label, filename_prefix, schedule, temporal_offset
+) VALUES (
+  1, 'new-source', 'newsource', '*/15 * * * *', 0
+);
+
+-- Link adapters to deployment
+INSERT INTO fetcher.deployment_adapters (deployments_id, adapters_id)
+VALUES (10, 42);
+```
+
+### Monitor deployment health
+```sql
+-- Find deployments not run recently
+SELECT label, schedule, last_deployed_datetime
+FROM fetcher.deployments
+WHERE is_active
+AND last_deployed_datetime < now() - interval '2 hours';
+```
+
+## Files
+
+- **scheduler.sql**: Cron expression validation and evaluation functions
+- **deployments.sql**: Tables, domain types, and scheduling functions

openaqdb/fetcher/deployment_data.sql

@@ -0,0 +1,110 @@
+  -- Data
+  SET search_path = fetcher, public;
+
+  INSERT INTO handlers (handlers_id, label, description, queue_name)
+  OVERRIDING SYSTEM VALUE
+  VALUES
+  (1, 'default', 'Fetch handler to use a default for deployments', 'default-fetcher');
+
+  INSERT INTO fetcher_clients (name, handler, description, authorization_method) VALUES
+    ('clarity', NULL, 'Transform client written for the Clarity API', 'API Key')
+  , ('air4thai', NULL, 'Custom transform client written for the air4thai API', NULL)
+  , ('bam', NULL, 'Custom transform client written for BAM data', NULL)
+  , ('data354', NULL, 'Custom transform client written for the data354 API', NULL)
+  , ('eea', NULL, 'Custom transform client written for EEA data', NULL)
+  , ('habitatmap', NULL, 'Custom transform client written for the HabitatMap API', NULL)
+  , ('senstate', NULL, 'Custom transform client written for the Senstate API', 'API Key')
+  , ('airqo', NULL, 'Custom transform client written for the Senstate API', NULL)
+  , ('openaq', NULL, 'Custom transform client written for the OpenAQ API', 'API Key')
+  , ('airnow', NULL, 'Custom transform client written for the OpenAQ API', NULL)
+  , ('london', NULL, 'Custom transform client written for the OpenAQ API', NULL)
+  , ('japan', NULL, 'Custom transform client written for the OpenAQ API', NULL)
+  , ('mexico', NULL, 'Custom transform client written for the OpenAQ API', NULL)
+  , ('hanoi', NULL, 'Custom transform client written for the OpenAQ API', NULL)
+  ON CONFLICT (name) DO UPDATE
+  SET description = EXCLUDED.description
+  , authorization_method = EXCLUDED.authorization_method;
+
+
+  -- now create an adapter with config for each of these
+  INSERT INTO adapters (fetcher_clients_id, providers_id, config)
+  SELECT a.fetcher_clients_id, p.providers_id, '{}'
+  FROM fetcher_clients a
+  JOIN public.providers p ON (lower(p.source_name) = lower(a.name))
+  ON CONFLICT DO NOTHING;
+
+  INSERT INTO adapters (fetcher_clients_id, providers_id, config)
+  SELECT ac.fetcher_clients_id
+  , p.providers_id
+  , '{}'
+  FROM (VALUES
+      ('hanoi', 'stateair_hanoi')
+    , ('mexico', 'sinaica mexico')
+    , ('airnow', 'airnow')
+    , ('london', 'london air quality network')
+    , ('japan', 'japan-soramame')
+  ) as v (fetcher_clients_name, source_name)
+  JOIN fetcher_clients ac ON (v.fetcher_clients_name = ac.name)
+  JOIN providers p ON (v.source_name = lower(p.source_name))
+  ON CONFLICT DO NOTHING;
+
+
+
+  -- create the deployments we currently have
+INSERT INTO deployments (deployments_id, label, description, temporal_offset, filename_prefix, schedule)
+OVERRIDING SYSTEM VALUE
+VALUES
+    (1, 'realtime', 'most government data', 0, 'realtime', '*/15 * * * *')
+  , (2, 'airnow cleanup', 'a deployment that rechecks for data from yesterday', 24, 'airnow', '0 * * * *')
+  , (3, 'london', 'breath london fetcher', 0, 'london', '*/15 * * * *')
+  , (4, 'acumar', '', 75, 'acumar', '*/15 * * * *')
+  , (5, 'japan', '', 0, 'japan', '*/15 * * * *')
+  , (6, 'mexico', '', 0, 'mexico', '*/15 * * * *')
+  , (7, 'hanoi', '', 12, 'hanoi', '*/15 * * * *')
+  , (8, 'clarity', '', 0, 'clarity', '0 * * * *')
+  , (9, 'senstate', '', 0, 'senstate', '*/5 * * * *')
+  , (10, 'testing-1min', 'this is one that should always fire unless its already been deployed for the current time', 0, 'senstate', '* * * * *')
+ ON CONFLICT DO NOTHING;
+
+
+  -- realtime should run them all
+  INSERT INTO deployment_adapters (deployments_id, adapters_id)
+  SELECT deployments_id, adapters_id
+  FROM deployments d, adapters
+  WHERE d.label ~* 'realtime'
+  ON CONFLICT DO NOTHING;
+
+  -- the rest should be one offs
+  INSERT INTO deployment_adapters (deployments_id, adapters_id)
+  SELECT d.deployments_id
+  , adapters_id
+  FROM adapters a
+  JOIN fetcher_clients c USING (fetcher_clients_id)
+  JOIN deployments d ON (c.name = d.label)
+  ON CONFLICT DO NOTHING;
+
+  -- airnow cleanup
+  INSERT INTO deployment_adapters (deployments_id, adapters_id)
+  SELECT deployments_id, adapters_id
+  FROM deployments d, adapters
+  JOIN fetcher_clients c USING (fetcher_clients_id)
+  WHERE d.label ~* 'cleanup'
+  AND c.name = 'airnow'
+  ON CONFLICT DO NOTHING
+  ;
+
+
+  INSERT INTO deployment_adapters (deployments_id, adapters_id)
+  SELECT deployments_id, adapters_id
+  FROM deployments d, adapters
+  JOIN fetcher_clients c USING (fetcher_clients_id)
+  WHERE d.label ~* 'testing'
+  AND c.name = 'senstate'
+  ON CONFLICT DO NOTHING
+  ;
+
+
+--SELECT * FROM fetcher.get_ready_deployments('2026-01-26 12:45:00');
+
+--SELECT * FROM fetcher.queue_deployments('2026-01-26 11:45:00');
+--SELECT * FROM fetcher.get_and_mark_queued_jobs();