Python API

The imo_vmdb package provides a Python API that can be used independently of any web framework or HTTP connection. It covers importing, normalising, cleaning up, exporting, and querying meteor observation data.

Database connection

class imo_vmdb.DBAdapter(config: dict[str, str])

Lightweight database connection adapter supporting SQLite, PostgreSQL, and MySQL.

A dedicated adapter is used instead of a heavier abstraction (e.g. SQLAlchemy) because the SQL in this project is straightforward and the only cross-backend difference is placeholder syntax. The adapter handles three concerns:

  • Dynamic driver selection — the DB-API 2.0 module (sqlite3, psycopg2, mysql.connector, …) is imported at runtime from config, so no specific driver is a hard dependency of the package.

  • SQLite initialisation — enables foreign-key enforcement via PRAGMA foreign_keys = ON, which SQLite disables by default.

  • Placeholder normalisation — see convert_stmt().

The default backend is sqlite3; a different module can be selected via the module key in config.

Parameters:

config –

Connection parameters. The optional module key names the DB-API 2.0 module to load (default: "sqlite3"); all remaining keys are forwarded verbatim to that module’s connect() call.

Typical keys per backend:

  • SQLite — database: path to the .db file.

  • PostgreSQL (module = psycopg2) — database, user, optionally host, password.

  • MySQL (module = pymysql) — database, user, optionally host, password, plus any driver-specific keys such as sql_mode or init_command.

These correspond directly to the [database] section of the imo-vmdb configuration file.

close() → None

Close the database connection.

commit() → None

Commit the current transaction.

convert_stmt(stmt: str) → str

Convert a SQL statement to the active backend’s dialect.

SQLite requires :name placeholders; this method rewrites %(name)s to :name and replaces %% with %. PostgreSQL (psycopg2) and MySQL (mysql-connector-python) accept %(name)s natively, so their statements are returned unchanged.

Parameters:

stmt – SQL statement using %(name)s-style named placeholders.

Returns:

Dialect-adjusted SQL statement.

cursor() → Any

Return a new database cursor.

Returns:

A new cursor object from the underlying DB-API 2.0 connection.

ping() → None

Verify that the database accepts a trivial query.

Issues a SELECT 1 and discards the result. Intended for liveness/readiness checks (e.g. /health endpoints).

Raises:

DBException – If the underlying driver raises any error.

year_expr(col: str) → str

Return a SQL fragment that extracts the year from a timestamp column.

Cross-dialect helper used by aggregate queries that need to group observations by year. Result type is integer.

Parameters:

col – SQL column reference (e.g. "r.period_start").

Returns:

Dialect-specific SQL expression yielding an integer year.

The keys of the config dict correspond directly to the [database] section of the configuration file (see Setup). Examples:

import imo_vmdb

# SQLite
db = imo_vmdb.DBAdapter({"database": "/path/to/vmdb.db"})

# PostgreSQL
db = imo_vmdb.DBAdapter({
    "module": "psycopg2",
    "database": "vmdb",
    "user": "vmdb",
    "host": "localhost",
})

# MySQL
db = imo_vmdb.DBAdapter({
    "module": "pymysql",
    "database": "vmdb",
    "user": "vmdb",
})
class imo_vmdb.DBException

Raised when a database operation fails.

Operations

imo_vmdb.cleanup(db_conn: DBAdapter, logger: Logger) → int

Remove all previously imported data, if any, while preserving normalized data in the database.

This function takes an existing database connection and a logger object as parameters. It removes all previously imported data from the database, leaving normalized data intact.

Parameters:

  • db_conn – An open database connection implementing DB-API 2.0.

  • logger (logging.Logger) – A logger object used to log errors, warnings, and additional information.

Returns:

An integer indicating the result of the operation. 0 for success, other values for errors.

Return type:

int

imo_vmdb.export_db(src_db_conn: DBAdapter, dst_conn: Connection) → None

Export normalized observations and reference data into a SQLite database.

Creates the full imo-vmdb schema on dst_conn (dropping any pre-existing tables) and copies all rows from the normalized and reference tables of src_db_conn into it. The raw imported_* tables are intentionally skipped — an exported database is a clean snapshot of normalized data that can be shared and used as the input of a fresh imo-vmdb install.

The destination connection is committed but not closed; its lifecycle is owned by the caller. This allows in-memory destinations (e.g. sqlite3.connect(":memory:")) to be inspected after export.

Parameters:

  • src_db_conn – Source database connection. May be SQLite, PostgreSQL, or MySQL.

  • dst_conn – Empty (or to-be-overwritten) SQLite destination connection. Must be a real sqlite3.Connection; other DB-API drivers are not supported as the destination.

imo_vmdb.export_table(db_conn: DBAdapter, table: str, reimport: bool = False) → tuple[list[str], list[tuple]]

Export all rows from a database table.

When reimport is True and table is 'shower', the result uses column names and date formats that are compatible with CSVImporter, so the exported CSV can be imported again without modification. For all other tables reimport has no effect.

Parameters:

  • db_conn – An open database connection implementing DB-API 2.0.

  • table – Name of the table to export. Must be one of the known tables: shower, radiant, obs_session, imported_session, imported_rate, imported_magnitude, rate, magnitude, rate_magnitude, magnitude_detail.

  • reimport – If True, export in re-import-compatible format where applicable.

Returns:

Tuple of (column_names, rows).

Raises:

ValueError – If table is not a known table name.

imo_vmdb.initdb(db_conn: DBAdapter, logger: Logger) → int

Initialize an empty database, removing all data if the database already exists.

This function takes an existing database connection and a logger object as parameters. It initializes an empty database, removing all data if the database already exists.

Parameters:

  • db_conn – An open database connection implementing DB-API 2.0.

  • logger (logging.Logger) – A logger object used to log errors, warnings, and additional information.

Returns:

An integer indicating the result of the operation. 0 for success, 1 for errors.

Return type:

int

imo_vmdb.normalize(db_conn: DBAdapter, logger: Logger) → int

Establish relationships between imported records and enrich observations with additional information.

This function takes an existing database connection and a logger object as parameters. It establishes relationships between the imported records in the database, enriching observations with additional information.

Parameters:

  • db_conn – An open database connection implementing DB-API 2.0.

  • logger (logging.Logger) – A logger object used to log errors, warnings, and additional information.

Returns:

An integer indicating the result of the operation. 0 for success, 1 for errors.

Return type:

int

Exporting a whole database

export_db() writes the normalized observations and reference data of a source database into an empty SQLite destination, using the same schema as a regular imo-vmdb database. The resulting file can be shared and re-used as the input of another imo-vmdb installation. The raw imported_* tables are intentionally excluded.

import sqlite3
import imo_vmdb

src = imo_vmdb.DBAdapter({"database": "/path/to/vmdb.db"})

# Export to a portable SQLite file:
dst = sqlite3.connect("/tmp/snapshot.sqlite")
imo_vmdb.export_db(src, dst)
dst.close()
class imo_vmdb.CSVImporter(db_conn: DBAdapter, logger: Logger, do_delete: bool = False, try_repair: bool = False, is_permissive: bool = False)

A class for importing CSV files of various types into a database.

The CSVImporter class allows to import CSV files into a database. You can specify whether to delete existing data, attempt data repair, or be permissive about non-critical data errors during the import.

Parameters:

  • db_conn – An existing database connection implementing DB-API 2.0.

  • logger (logging.Logger) – A logger object used to log errors, warnings, and additional information.

  • do_delete (bool) – If True, delete existing data before importing. Default is False.

  • try_repair (bool) – If True, attempt data repair during import. Default is False.

  • is_permissive (bool) – If True, be permissive about non-critical data errors. Default is False.

run(file_list: list[str]) → None

Import CSV files specified in the files_list into the database.

This method imports CSV files into the database, with options to delete existing data, attempt data repair, and be permissive about non-critical data errors. After running this method, you can check the has_errors, counter_read, and counter_write properties of this object to determine the import result.

Parameters:

file_list (list of str) – A list of file paths to CSV files for import.

Service classes

Each service class wraps a DBAdapter connection and exposes the queries available for one entity. The query method returns the matching list (with optional pagination/total), by_id / by_code return a single record (or None).

class imo_vmdb.RateService(db_conn: DBAdapter)

Service for rate observation queries.

Parameters:

db_conn – An open DBAdapter connection.

by_id(rate_id: int) → Rate | None

Return a single rate observation by ID, or None if not found.

query(f: RateFilter) → Rates

Return rate observations matching f.

Parameters:

f – A RateFilter specifying filter criteria and includes.

Returns:

A Rates instance. sessions and magnitudes are only set when the corresponding flags on f are True. total is set when with_total is True or when pagination is in use.

class imo_vmdb.MagnitudeService(db_conn: DBAdapter)

Service for magnitude observation queries.

Parameters:

db_conn – An open DBAdapter connection.

by_id(magn_id: int) → Magnitude | None

Return a single magnitude observation by ID, or None if not found.

query(f: MagnitudeFilter) → Magnitudes

Return magnitude observations matching f.

Parameters:

f – A MagnitudeFilter specifying filter criteria and includes.

Returns:

A Magnitudes instance. sessions and magnitudes are only set when the corresponding flags on f are True. total is set when with_total is True or when pagination is in use.

class imo_vmdb.SessionService(db_conn: DBAdapter)

Service for observation session queries.

Parameters:

db_conn – An open DBAdapter connection.

by_id(session_id: int) → Session | None

Return a single session by ID, or None if not found.

query(f: SessionFilter) → Sessions

Return sessions matching f.

class imo_vmdb.ShowerService(db_conn: DBAdapter)

Service for meteor shower reference data and radiants.

Parameters:

db_conn – An open DBAdapter connection.

active(on_date: date) → list[Shower]

Return all showers whose activity period covers on_date.

Handles year-wrapping showers (e.g. start in December, end in January) by treating the period as inclusive on both ends.

Parameters:

on_date – Calendar date to test against each shower’s start_* / end_* fields. Year is ignored.

Returns:

List of matching Shower instances.

by_code(iau_code: str) → Shower | None

Return the shower with the given IAU code, or None if unknown.

query() → list[Shower]

Return all meteor showers from the catalogue, ordered by IAU code.

radiants(iau_code: str) → list[Radiant]

Return all radiant entries for iau_code, ordered by (month, day).

class imo_vmdb.StatsService(db_conn: DBAdapter)

Aggregate statistics over the observation database.

All by_* methods accept an optional period_start/period_end filter (ISO date strings) that restricts the rate and magnitude tables before aggregation. Sessions are not period-filtered for meta and by_country to keep the semantics simple (“how many sessions are in the database”).

Parameters:

db_conn – An open DBAdapter connection.

by_country(period_start: str | None = None, period_end: str | None = None) → list[CountryStat]

Return per-country counts of sessions, rates and magnitudes.

by_shower(period_start: str | None = None, period_end: str | None = None) → list[ShowerStat]

Return per-shower counts of rates and magnitudes.

by_year(period_start: str | None = None, period_end: str | None = None) → list[YearStat]

Return per-year counts of rates and magnitudes.

meta() → StatsMeta

Return overall counts and the covered date range.

Filter types

class imo_vmdb.RateFilter(showers: list[str] = <factory>, period_start: str | None = None, period_end: str | None = None, sl_min: float | None = None, sl_max: float | None = None, lim_magn_min: float | None = None, lim_magn_max: float | None = None, sun_alt_max: float | None = None, moon_alt_max: float | None = None, session_ids: list[int] = <factory>, rate_ids: list[int] = <factory>, include_sessions: bool = False, include_magnitudes: bool = False, limit: int | None = None, offset: int | None = None, order_by: str | None = None, order: str | None = None, with_total: bool = False)

Filter criteria for rate observation queries.

Parameters:

  • showers – IAU shower codes to include; use 'SPO' for sporadics.

  • period_start – Include only observations starting on or after this date (YYYY-MM-DD).

  • period_end – Include only observations ending on or before this date (YYYY-MM-DD).

  • sl_min – Minimum solar longitude (start of period).

  • sl_max – Maximum solar longitude (end of period).

  • lim_magn_min – Minimum limiting magnitude.

  • lim_magn_max – Maximum limiting magnitude.

  • sun_alt_max – Maximum sun altitude in degrees.

  • moon_alt_max – Maximum moon altitude in degrees.

  • session_ids – Restrict to specific session IDs.

  • rate_ids – Restrict to specific rate record IDs.

  • include_sessions – If True, include a sessions list in the result.

  • include_magnitudes – If True, include a magnitudes list with the per-class magnitude-distribution detail rows (from magnitude_detail) linked to each rate observation.

class imo_vmdb.MagnitudeFilter(showers: list[str] = <factory>, period_start: str | None = None, period_end: str | None = None, sl_min: float | None = None, sl_max: float | None = None, lim_magn_min: float | None = None, lim_magn_max: float | None = None, session_ids: list[int] = <factory>, magn_ids: list[int] = <factory>, include_sessions: bool = False, include_magnitudes: bool = False, limit: int | None = None, offset: int | None = None, order_by: str | None = None, order: str | None = None, with_total: bool = False)

Filter criteria for magnitude observation queries.

Parameters:

  • showers – IAU shower codes to include; use 'SPO' for sporadics.

  • period_start – Include only observations starting on or after this date (YYYY-MM-DD).

  • period_end – Include only observations ending on or before this date (YYYY-MM-DD).

  • sl_min – Minimum solar longitude (start of period).

  • sl_max – Maximum solar longitude (end of period).

  • lim_magn_min – Minimum limiting magnitude.

  • lim_magn_max – Maximum limiting magnitude.

  • session_ids – Restrict to specific session IDs.

  • magn_ids – Restrict to specific magnitude record IDs.

  • include_sessions – If True, include a sessions list in the result.

  • include_magnitudes – If True, include a magnitudes list with the per-class magnitude-distribution detail rows (from magnitude_detail) for each magnitude observation.

class imo_vmdb.SessionFilter(observer_ids: list[int] = <factory>, period_start: str | None = None, period_end: str | None = None, limit: int | None = None, offset: int | None = None, order_by: str | None = None, order: str | None = None, with_total: bool = False)

Filter criteria for session queries.

Parameters:

  • observer_ids – Restrict to specific observer IDs.

  • period_start – Include only sessions with at least one rate or magnitude observation starting on or after this date.

  • period_end – Include only sessions with at least one rate or magnitude observation ending on or before this date.

  • limit – Maximum number of sessions to return.

  • offset – Number of leading sessions to skip.

  • order_by – Sort column (whitelist: id, country, observer_id).

  • order – "asc" or "desc".

  • with_total – If True, populate Sessions.total.

Result types

class imo_vmdb.Shower(iau_code: str, name: str, start_month: int, start_day: int, end_month: int, end_day: int, peak_month: int | None, peak_day: int | None, ra: float | None, dec: float | None, v: float | None, r: float | None, zhr: float | None)

Single meteor shower from the reference catalogue.

See Field Reference for field descriptions.

class imo_vmdb.Session(id: int, longitude: float, latitude: float, elevation: float, country: str, city: str, observer_id: int | None, observer_name: str | None)

Observation session linked to rate or magnitude observations.

See Field Reference for field descriptions.

class imo_vmdb.Rate(id: int, shower: str | None, period_start: str, period_end: str, sl_start: float, sl_end: float, session_id: int, freq: int, lim_mag: float, t_eff: float, f: float, sidereal_time: float, sun_alt: float, sun_az: float, moon_alt: float, moon_az: float, moon_illum: float, field_alt: float | None, field_az: float | None, rad_alt: float | None, rad_az: float | None, magn_id: int | None)

Single normalised rate observation returned by RateService.query().

See Field Reference for field descriptions.

magn_id links to the associated Magnitude observation via the rate_magnitude join table, or is None when no matching magnitude observation exists.

class imo_vmdb.Magnitude(id: int, shower: str | None, period_start: str, period_end: str, sl_start: float, sl_end: float, session_id: int, freq: int, mean: float, lim_mag: float | None)

Single normalised magnitude observation returned by MagnitudeService.query().

See Field Reference for field descriptions.

class imo_vmdb.MagnitudeDetail(id: int, magn: int, freq: float)

Per-class frequency entry for a magnitude observation.

See Field Reference for field descriptions.

class imo_vmdb.Radiant(shower: str, month: int, day: int, ra: float, dec: float)

Radiant position of a meteor shower at a given calendar day.

class imo_vmdb.Rates(observations: list[Rate], sessions: list[Session] | None = None, magnitudes: list[MagnitudeDetail] | None = None, total: int | None = None)

Return value of RateService.query().

Always contains an observations list of Rate instances. sessions (list of Session) is only set when RateFilter.include_sessions is True. magnitudes (list of MagnitudeDetail) is only set when RateFilter.include_magnitudes is True. total is the unpaginated row count and is only set when RateFilter.with_total is True (or when limit/offset are used).

class imo_vmdb.Magnitudes(observations: list[Magnitude], sessions: list[Session] | None = None, magnitudes: list[MagnitudeDetail] | None = None, total: int | None = None)

Return value of MagnitudeService.query().

Always contains an observations list of Magnitude instances. sessions (list of Session) is only set when MagnitudeFilter.include_sessions is True. magnitudes (list of MagnitudeDetail) is only set when MagnitudeFilter.include_magnitudes is True. total is the unpaginated row count and is only set when MagnitudeFilter.with_total is True (or when limit/offset are used).

class imo_vmdb.Sessions(observations: list[Session], total: int | None = None)

Return value of SessionService.query().

class imo_vmdb.StatsMeta(sessions: int, rates: int, magnitudes: int, period_start: str | None, period_end: str | None)

Database scope summary returned by StatsService.meta().

class imo_vmdb.ShowerStat(shower: str | None, rates: int, magnitudes: int)

Per-shower aggregate counts. shower is None for sporadics.

class imo_vmdb.CountryStat(country: str, sessions: int, rates: int, magnitudes: int)

Per-country aggregate counts.

class imo_vmdb.YearStat(year: int, rates: int, magnitudes: int)

Per-year aggregate counts.

WSGI deployment

The web UI and REST API can be hosted under any WSGI server using the public app factory imo_vmdb.httpd.wsgi_app. Configuration is read from the IMO_VMDB_CONFIG environment variable (path to an INI file) or directly from IMO_VMDB_* variables (see Setup).

imo_vmdb.httpd.wsgi_app()

WSGI app factory for Gunicorn --factory mode.

Configuration is read from the IMO_VMDB_CONFIG environment variable (path to an INI file) or from IMO_VMDB_DATABASE_* etc. directly. Host and port are controlled by Gunicorn’s --bind flag.

The Web UI is opt-in: set IMO_VMDB_WEBSERVER_ENABLE_WEBUI=true to enable it. By default only the REST API is served.

Example:

gunicorn --workers 1 --threads 4 "imo_vmdb.httpd:wsgi_app()"

Warning

Always use --workers 1. The JobManager stores job state in-process; multiple workers would make jobs invisible across processes, breaking status polling and log streaming.

Example using Gunicorn (install with pip install "imo-vmdb[web]"):

# With a config file:
IMO_VMDB_CONFIG=config.ini \
    gunicorn --workers 1 --threads 4 \
    --bind 127.0.0.1:8000 "imo_vmdb.httpd:wsgi_app()"

# Without a config file:
IMO_VMDB_DATABASE_DATABASE=./vmdb.db \
    gunicorn --workers 1 --threads 4 \
    --bind 127.0.0.1:8000 "imo_vmdb.httpd:wsgi_app()"

Warning

Always use --workers 1. The job manager stores job state in-process; multiple workers would make jobs invisible across processes, breaking status polling and log streaming.