A minimal, standards-compliant TAP 1.1 (Table Access Protocol) service for astronomical catalog queries. Single Python file, SQLite backend, no external VO dependencies.

TAP is the IVOA standard protocol for SQL-like access to astronomical catalogs. Every major VO tool — TOPCAT, Aladin, pyvo, astroquery — speaks TAP natively. Publishing a catalog as a TAP service makes it immediately discoverable and queryable by the entire VO ecosystem without custom client code.

This implementation exists to answer: "what is the smallest possible TAP-compliant service?" It is a working reference, a local dev sandbox, and a scaffold for building production services. The full spec (ADQL, VOTable, UWS, VOSI, DALI) is large; this strips it to the mandatory surface with zero infrastructure overhead.

Trade-offs made:

• Python 3.11+
• uv (or pip)

# create venv and install
uv venv
source .venv/bin/activate
uv pip install flask

# start the service (creates tap.db on first run)
python tap_server.py
# → TAP service at http://localhost:8081/tap

# top 5 brightest stars
curl http://localhost:8081/tap/sync \
  -d "LANG=ADQL&QUERY=SELECT TOP 5 name,ra,dec,mag_v FROM catalog.stars ORDER BY mag_v"

# filter by spectral type
curl http://localhost:8081/tap/sync \
  -d "LANG=ADQL&QUERY=SELECT name,sptype,mag_v FROM catalog.stars WHERE sptype LIKE 'B%'"

# list all tables
curl http://localhost:8081/tap/sync \
  -d "LANG=ADQL&QUERY=SELECT table_name,description FROM TAP_SCHEMA.tables"

# column metadata for the catalog
curl http://localhost:8081/tap/sync \
  -d "LANG=ADQL&QUERY=SELECT column_name,datatype,unit,ucd FROM TAP_SCHEMA.columns WHERE table_name='catalog.stars'"

import pyvo

svc = pyvo.dal.TAPService("http://localhost:8081/tap")

# browse schema
print(svc.tables["catalog.stars"].columns)

# query
results = svc.search("SELECT * FROM catalog.stars WHERE mag_v < 0.5")
print(results.to_table())

TOPCAT:  File → Open TAP service → http://localhost:8081/tap
Firefly: TAP URL → http://localhost:8081/tap   (or http://localhost:8081 — both work)

# submit job (returns Location header with job URL)
JOB=$(curl -si http://localhost:8081/tap/async \
  -d "LANG=ADQL&QUERY=SELECT * FROM catalog.stars" \
  | grep -i location | awk '{print $2}' | tr -d '\r')

# trigger execution
curl -X POST "${JOB}/phase" -d "PHASE=RUN"

# poll until COMPLETED
curl "${JOB}/phase"

# retrieve result
curl "${JOB}/results/result"

All endpoints are registered at both /tap/<path> and /<path> (root alias). Clients configured with or without the /tap base suffix both work.

The sync endpoint accepts the standard REQUEST=doQuery parameter (required by the TAP DAL spec). If REQUEST is absent it defaults to doQuery for backward compatibility. Any other REQUEST value returns an error VOTable.

catalog.stars — 20 brightest stars (IAU naming, J2000 coordinates):

Client
  │
  ├─ POST /tap/sync ──→ sync()
  │                       ├─ parse LANG, QUERY, MAXREC
  │                       ├─ adql_to_sqlite(query)   ← regex translator
  │                       ├─ SQLite execute
  │                       └─ rows_to_votable()  ──→  VOTable XML response
  │
  └─ POST /tap/async ──→ async_list()
                          ├─ create JOBS[id] = {phase: PENDING, ...}
                          ├─ if PHASE=RUN: thread(_run_job)
                          └─ 303 → /tap/async/<id>

_run_job(id) [thread]
  ├─ run_query()
  ├─ rows_to_votable()
  └─ JOBS[id].phase = COMPLETED | ERROR

Regex-based. Handles:

• SELECT TOP N ... → SELECT ... LIMIT N
• TAP_SCHEMA.tables → tap_schema_tables (and all other TAP_SCHEMA tables)
• catalog.stars → catalog_stars
• INNER JOIN across TAP_SCHEMA tables — word-boundary substitution correctly rewrites qualified column refs (TAP_SCHEMA.tables.schema_name → tap_schema_tables.schema_name) in ON clauses

Does not handle: geometry functions (CONTAINS, CIRCLE, POINT), OUTER JOIN, correlated subqueries, UPLOAD table references, or non-ADQL languages.

Built as a string. Type inference: inspects first non-null Python value per column to choose int / double / char. Produces valid VOTable 1.4 with QUERY_STATUS INFO element (OK / ERROR / OVERFLOW).

All five required tables exist in SQLite and are queryable via ADQL: TAP_SCHEMA.schemas, TAP_SCHEMA.tables, TAP_SCHEMA.columns, TAP_SCHEMA.keys, TAP_SCHEMA.key_columns. Seeded in init_db() on first run.

In-memory dict JOBS. Each job: {phase, lang, query, maxrec, result, error}. Phase lifecycle: PENDING → QUEUED → EXECUTING → COMPLETED | ERROR. Jobs run in daemon threads. No persistence — lost on process restart.

tap_server.py    — entire service (~520 lines)
tap.db           — SQLite database (auto-created; delete to reset)
requirements.txt — flask>=3.0
CLAUDE.md        — developer/agent guide, compliance status, full roadmap
README.md        — this file

See CLAUDE.md for the full prioritised roadmap. Short version:

1. Output formats — CSV, TSV (RESPONSEFORMAT parameter)
2. TABLE UPLOAD — client-uploaded VOTable as temp query target
3. Persistent async jobs — store in DB, survive restarts
4. Geometry — CONTAINS/CIRCLE/BOX ADQL functions via SpatiaLite or bounding box
5. Production server — gunicorn -w 4 tap_server:app
6. PostgreSQL backend — replace SQLite for concurrent access at scale
7. Proper ADQL parser — ANTLR4 grammar, SQL injection protection
8. IVOA Registry record — VOResource XML for VO discoverability
9. ObsCore table — standard data model for observational metadata
10. Auth (SSO) — X.509 / cookie per IVOA SSO standard

• TAP 1.1 — Table Access Protocol
• ADQL 2.1 — Astronomical Data Query Language
• UWS 1.1 — Universal Worker Service (async jobs)
• VOSI 1.1 — VO Support Interfaces
• VOTable 1.4 — result serialization format
• DALI 1.1 — Data Access Layer Interface rules