update gitignore file
This commit is contained in:
@@ -14,3 +14,4 @@ dbsync.log.*
|
||||
/log
|
||||
venv
|
||||
/temp
|
||||
/archived
|
||||
@@ -1,38 +0,0 @@
|
||||
# SQLite -> PostgreSQL virtual table (minimal)
|
||||
|
||||
This directory contains a minimal SQLite extension that implements a read-only virtual table backed directly by PostgreSQL using libpq.
|
||||
|
||||
Build
|
||||
-----
|
||||
You need `libpq` (Postgres client) and a C compiler. On Debian/Ubuntu:
|
||||
|
||||
```bash
|
||||
sudo apt-get install libpq-dev build-essential
|
||||
make
|
||||
```
|
||||
|
||||
This produces `pg_vtab.so`.
|
||||
|
||||
Usage
|
||||
-----
|
||||
Start `sqlite3`, load the extension and create a virtual table pointing at a Postgres table:
|
||||
|
||||
```sql
|
||||
.load ./pg_vtab.so
|
||||
CREATE VIRTUAL TABLE remote_tbl USING pg_vtab('host=localhost dbname=mydb user=me password=secret', 'public.mytable');
|
||||
SELECT * FROM remote_tbl LIMIT 10;
|
||||
```
|
||||
|
||||
Limitations
|
||||
-----------
|
||||
- Minimal proof-of-concept: supports only full table scans.
|
||||
- All columns are presented as `TEXT`.
|
||||
- No constraint pushdown or type mapping sophistication.
|
||||
- Error handling and SQL injection hygiene are minimal; do not use on untrusted inputs without review.
|
||||
|
||||
Next steps (optional)
|
||||
---------------------
|
||||
- Add support for WHERE constraints and indexed access (`xBestIndex`).
|
||||
- Map Postgres types more accurately to SQLite storage classes.
|
||||
- Implement server-side cursors for large tables.
|
||||
- Harden error reporting and resource cleanup.
|
||||
@@ -1,22 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import os
|
||||
import json
|
||||
import requests
|
||||
import time
|
||||
|
||||
def download_json_to_file(api_url, outputdir, output_file, lprint):
|
||||
"""
|
||||
Downloads JSON data from the given API URL and saves it to the specified file.
|
||||
"""
|
||||
try:
|
||||
response = requests.get(api_url)
|
||||
response.raise_for_status() # Raise an error for bad status codes
|
||||
data = response.json()
|
||||
with open(f"{outputdir}{output_file}", 'w') as f:
|
||||
json.dump(data, f, indent=4)
|
||||
#print(f"JSON data saved to {outputdir}{output_file}")
|
||||
time.sleep(0.5) # Sleep for 0.5 second to avoid overwhelming the server
|
||||
except Exception as e:
|
||||
lprint.logprint("info", f"Downloadinng {outputdir}{output_file} for series ID: {output_file.replace('.json', '')} failed: {e}")
|
||||
|
||||
@@ -1,259 +0,0 @@
|
||||
/*
|
||||
* Minimal SQLite virtual table module backed by PostgreSQL via libpq.
|
||||
*
|
||||
* Usage (after building as shared object):
|
||||
* sqlite3> .load ./pg_vtab.so
|
||||
* sqlite> CREATE VIRTUAL TABLE t1 USING pg_vtab('host=... dbname=... user=... password=...', 'schema.table');
|
||||
* sqlite> SELECT * FROM t1 LIMIT 10;
|
||||
*
|
||||
* Notes:
|
||||
* - This is a minimal read-only implementation.
|
||||
* - It only supports full table scans (no constraint pushdown).
|
||||
* - All values are returned as TEXT.
|
||||
*/
|
||||
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <assert.h>
|
||||
|
||||
#include <libpq-fe.h>
|
||||
#include <sqlite3.h>
|
||||
|
||||
/* vtab object */
|
||||
typedef struct PgVTab {
|
||||
sqlite3_vtab base; /* must be first */
|
||||
char *conninfo;
|
||||
char *schema_table; /* qualified table name */
|
||||
int ncol;
|
||||
char **colnames;
|
||||
} PgVTab;
|
||||
|
||||
/* cursor object */
|
||||
typedef struct PgVTabCursor {
|
||||
sqlite3_vtab_cursor base; /* must be first */
|
||||
PGconn *pgconn;
|
||||
PGresult *res;
|
||||
int row; /* current row index in res */
|
||||
sqlite3_int64 rowid; /* generated rowid */
|
||||
} PgVTabCursor;
|
||||
|
||||
/* forward declarations */
|
||||
static int pgConnect(sqlite3 *db, void *pAux, int argc, const char *const *argv, sqlite3_vtab **ppVTab, char **pzErr);
|
||||
static int pgDisconnect(sqlite3_vtab *pVTab);
|
||||
static int pgBestIndex(sqlite3_vtab *pVTab, sqlite3_index_info *pIdx);
|
||||
static int pgOpen(sqlite3_vtab *pVTab, sqlite3_vtab_cursor **ppCursor);
|
||||
static int pgClose(sqlite3_vtab_cursor *cur);
|
||||
static int pgFilter(sqlite3_vtab_cursor *pVtabCursor, int idxNum, const char *idxStr, int argc, sqlite3_value **argv);
|
||||
static int pgNext(sqlite3_vtab_cursor *cur);
|
||||
static int pgEof(sqlite3_vtab_cursor *cur);
|
||||
static int pgColumn(sqlite3_vtab_cursor *cur, sqlite3_context *ctx, int i);
|
||||
static int pgRowid(sqlite3_vtab_cursor *cur, sqlite3_int64 *pRowid);
|
||||
|
||||
static sqlite3_module PgModule = {
|
||||
0, /* iVersion */
|
||||
pgConnect,
|
||||
pgConnect, /* xCreate == xConnect for this module */
|
||||
pgBestIndex,
|
||||
pgDisconnect,
|
||||
pgDisconnect, /* xDestroy == xDisconnect */
|
||||
pgOpen,
|
||||
pgClose,
|
||||
pgFilter,
|
||||
pgNext,
|
||||
pgEof,
|
||||
pgColumn,
|
||||
pgRowid,
|
||||
NULL, NULL, NULL, NULL, NULL, NULL, NULL
|
||||
};
|
||||
|
||||
/* helper: strdup-like */
|
||||
static char *str_dup(const char *s){
|
||||
if(!s) return NULL;
|
||||
size_t n = strlen(s)+1;
|
||||
char *r = (char*)malloc(n);
|
||||
if(r) memcpy(r,s,n);
|
||||
return r;
|
||||
}
|
||||
|
||||
/* read column names from POSTGRES by executing SELECT * FROM table LIMIT 0 */
|
||||
static int fetch_pg_columns(const char *conninfo, const char *schema_table, int *pncol, char ***pnames, char **pErr){
|
||||
PGconn *conn = PQconnectdb(conninfo);
|
||||
if (PQstatus(conn) != CONNECTION_OK) {
|
||||
if(pErr) *pErr = str_dup(PQerrorMessage(conn));
|
||||
PQfinish(conn);
|
||||
return SQLITE_ERROR;
|
||||
}
|
||||
char q[1024];
|
||||
snprintf(q, sizeof(q), "SELECT * FROM %s LIMIT 0", schema_table);
|
||||
PGresult *res = PQexec(conn, q);
|
||||
if (PQresultStatus(res) != PGRES_TUPLES_OK && PQresultStatus(res) != PGRES_COMMAND_OK) {
|
||||
if(pErr) *pErr = str_dup(PQerrorMessage(conn));
|
||||
PQclear(res);
|
||||
PQfinish(conn);
|
||||
return SQLITE_ERROR;
|
||||
}
|
||||
int n = PQnfields(res);
|
||||
char **names = (char**)malloc(sizeof(char*)*n);
|
||||
for(int i=0;i<n;i++){
|
||||
names[i] = str_dup(PQfname(res,i));
|
||||
}
|
||||
PQclear(res);
|
||||
PQfinish(conn);
|
||||
*pncol = n;
|
||||
*pnames = names;
|
||||
return SQLITE_OK;
|
||||
}
|
||||
|
||||
static int pgConnect(sqlite3 *db, void *pAux, int argc, const char *const *argv, sqlite3_vtab **ppVTab, char **pzErr){
|
||||
/* argv: module name, db name, conninfo, schema.table */
|
||||
if(argc < 4){
|
||||
if(pzErr) *pzErr = str_dup("pg_vtab requires connection string and qualified table name");
|
||||
return SQLITE_ERROR;
|
||||
}
|
||||
const char *conninfo = argv[2];
|
||||
const char *schema_table = argv[3];
|
||||
|
||||
int ncol = 0;
|
||||
char **names = NULL;
|
||||
int rc = fetch_pg_columns(conninfo, schema_table, &ncol, &names, pzErr);
|
||||
if(rc != SQLITE_OK) return rc;
|
||||
|
||||
/* build CREATE TABLE declaration with TEXT columns */
|
||||
size_t needed = 64;
|
||||
for(int i=0;i<ncol;i++) needed += strlen(names[i]) + 8;
|
||||
char *decl = (char*)malloc(needed);
|
||||
if(!decl){
|
||||
if(pzErr) *pzErr = str_dup("out of memory");
|
||||
return SQLITE_NOMEM;
|
||||
}
|
||||
strcpy(decl, "CREATE TABLE x(");
|
||||
for(int i=0;i<ncol;i++){
|
||||
if(i) strcat(decl, ",");
|
||||
strcat(decl, names[i]);
|
||||
strcat(decl, " TEXT");
|
||||
}
|
||||
strcat(decl, ")");
|
||||
|
||||
rc = sqlite3_declare_vtab(db, decl);
|
||||
free(decl);
|
||||
if(rc != SQLITE_OK){
|
||||
if(pzErr) *pzErr = str_dup("sqlite3_declare_vtab failed");
|
||||
for(int i=0;i<ncol;i++) free(names[i]); free(names);
|
||||
return rc;
|
||||
}
|
||||
|
||||
PgVTab *vtab = (PgVTab*)sqlite3_malloc(sizeof(PgVTab));
|
||||
if(!vtab) return SQLITE_NOMEM;
|
||||
memset(vtab,0,sizeof(PgVTab));
|
||||
vtab->conninfo = str_dup(conninfo);
|
||||
vtab->schema_table = str_dup(schema_table);
|
||||
vtab->ncol = ncol;
|
||||
vtab->colnames = names;
|
||||
|
||||
*ppVTab = (sqlite3_vtab*)vtab;
|
||||
return SQLITE_OK;
|
||||
}
|
||||
|
||||
static int pgDisconnect(sqlite3_vtab *pVTab){
|
||||
PgVTab *v = (PgVTab*)pVTab;
|
||||
if(!v) return SQLITE_OK;
|
||||
if(v->conninfo) free(v->conninfo);
|
||||
if(v->schema_table) free(v->schema_table);
|
||||
for(int i=0;i<v->ncol;i++) free(v->colnames[i]);
|
||||
if(v->colnames) free(v->colnames);
|
||||
sqlite3_free(v);
|
||||
return SQLITE_OK;
|
||||
}
|
||||
|
||||
static int pgBestIndex(sqlite3_vtab *pVTab, sqlite3_index_info *pIdx){
|
||||
/* do not support indexes or constraints; full table scan */
|
||||
pIdx->estimatedCost = 1e6;
|
||||
return SQLITE_OK;
|
||||
}
|
||||
|
||||
static int pgOpen(sqlite3_vtab *pVTab, sqlite3_vtab_cursor **ppCursor){
|
||||
PgVTabCursor *cur = (PgVTabCursor*)sqlite3_malloc(sizeof(PgVTabCursor));
|
||||
if(!cur) return SQLITE_NOMEM;
|
||||
memset(cur,0,sizeof(PgVTabCursor));
|
||||
*ppCursor = &cur->base;
|
||||
cur->pgconn = NULL;
|
||||
cur->res = NULL;
|
||||
cur->row = 0;
|
||||
cur->rowid = 0;
|
||||
return SQLITE_OK;
|
||||
}
|
||||
|
||||
static int pgClose(sqlite3_vtab_cursor *curp){
|
||||
PgVTabCursor *cur = (PgVTabCursor*)curp;
|
||||
if(cur->res) PQclear(cur->res);
|
||||
if(cur->pgconn) PQfinish(cur->pgconn);
|
||||
sqlite3_free(cur);
|
||||
return SQLITE_OK;
|
||||
}
|
||||
|
||||
static int pgFilter(sqlite3_vtab_cursor *pVtabCursor, int idxNum, const char *idxStr, int argc, sqlite3_value **argv){
|
||||
PgVTabCursor *cur = (PgVTabCursor*)pVtabCursor;
|
||||
PgVTab *vtab = (PgVTab*)(pVtabCursor->pVtab);
|
||||
/* connect to pg */
|
||||
if(cur->res){ PQclear(cur->res); cur->res = NULL; }
|
||||
if(cur->pgconn){ PQfinish(cur->pgconn); cur->pgconn = NULL; }
|
||||
cur->pgconn = PQconnectdb(vtab->conninfo);
|
||||
if(PQstatus(cur->pgconn) != CONNECTION_OK){
|
||||
/* can't report error via sqlite3_errmsg here, return error */
|
||||
return SQLITE_ERROR;
|
||||
}
|
||||
|
||||
/* simple select all */
|
||||
char q[1024];
|
||||
snprintf(q, sizeof(q), "SELECT * FROM %s", vtab->schema_table);
|
||||
cur->res = PQexec(cur->pgconn, q);
|
||||
if(PQresultStatus(cur->res) != PGRES_TUPLES_OK && PQresultStatus(cur->res) != PGRES_COMMAND_OK){
|
||||
PQclear(cur->res); cur->res = NULL; return SQLITE_ERROR;
|
||||
}
|
||||
cur->row = 0;
|
||||
cur->rowid = 0;
|
||||
return SQLITE_OK;
|
||||
}
|
||||
|
||||
static int pgNext(sqlite3_vtab_cursor *curp){
|
||||
PgVTabCursor *cur = (PgVTabCursor*)curp;
|
||||
cur->row++;
|
||||
cur->rowid++;
|
||||
return SQLITE_OK;
|
||||
}
|
||||
|
||||
static int pgEof(sqlite3_vtab_cursor *curp){
|
||||
PgVTabCursor *cur = (PgVTabCursor*)curp;
|
||||
if(!cur->res) return 1;
|
||||
int nt = PQntuples(cur->res);
|
||||
return cur->row >= nt;
|
||||
}
|
||||
|
||||
static int pgColumn(sqlite3_vtab_cursor *curp, sqlite3_context *ctx, int i){
|
||||
PgVTabCursor *cur = (PgVTabCursor*)curp;
|
||||
if(!cur->res) { sqlite3_result_null(ctx); return SQLITE_OK; }
|
||||
if(PQgetisnull(cur->res, cur->row, i)){
|
||||
sqlite3_result_null(ctx);
|
||||
} else {
|
||||
const char *val = PQgetvalue(cur->res, cur->row, i);
|
||||
sqlite3_result_text(ctx, val, -1, SQLITE_TRANSIENT);
|
||||
}
|
||||
return SQLITE_OK;
|
||||
}
|
||||
|
||||
static int pgRowid(sqlite3_vtab_cursor *curp, sqlite3_int64 *pRowid){
|
||||
PgVTabCursor *cur = (PgVTabCursor*)curp;
|
||||
*pRowid = cur->rowid;
|
||||
return SQLITE_OK;
|
||||
}
|
||||
|
||||
/* Entry point called by sqlite3 when loading extension */
|
||||
int sqlite3_pg_vtab_init(sqlite3 *db, char **pzErrMsg, const sqlite3_api_routines *pApi){
|
||||
SQLITE_EXTENSION_INIT2(pApi)}
|
||||
int rc = sqlite3_create_module(db, "pg_vtab", &PgModule, NULL);
|
||||
if(rc != SQLITE_OK){
|
||||
if(pzErrMsg) *pzErrMsg = str_dup("failed to register pg_vtab module");
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
Executable
+3
@@ -0,0 +1,3 @@
|
||||
#!/bin/bash
|
||||
cd /srv/Projects/new_dbsync
|
||||
/srv/Projects/new_dbsync/update_json.py >> /home/wjones/update_mongo.log
|
||||
-193
@@ -1,193 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Simple utility to copy a single table from SQLite to PostgreSQL.
|
||||
|
||||
Usage example:
|
||||
python sqlite2pgsql /path/to/sqlite.db "host=... dbname=... user=... password=..." mytable
|
||||
|
||||
Implements `convert_table_sqlite_to_pg(sqlite_path, pg_dsn, table_name, create_table=True, batch_size=1000)`
|
||||
which copies schema (basic types) and data in batches.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import sqlite3
|
||||
from typing import List, Tuple
|
||||
|
||||
import psycopg2
|
||||
from psycopg2 import sql
|
||||
|
||||
LOG = logging.getLogger("sqlite2pgsql")
|
||||
|
||||
|
||||
def _map_sqlite_type_to_pg(sqlite_type: str) -> str:
|
||||
t = (sqlite_type or "").upper()
|
||||
if "INT" in t:
|
||||
return "INTEGER"
|
||||
if any(x in t for x in ("CHAR", "CLOB", "TEXT")):
|
||||
return "TEXT"
|
||||
if "BLOB" in t:
|
||||
return "BYTEA"
|
||||
if any(x in t for x in ("REAL", "FLOA", "DOUB")):
|
||||
return "REAL"
|
||||
if "NUM" in t or "DEC" in t:
|
||||
return "NUMERIC"
|
||||
# default fallback
|
||||
return "TEXT"
|
||||
|
||||
|
||||
def _get_table_info_sqlite(conn: sqlite3.Connection, table: str) -> List[Tuple]:
|
||||
cur = conn.execute(f"PRAGMA table_info('{table}')")
|
||||
rows = cur.fetchall()
|
||||
return rows
|
||||
|
||||
|
||||
def list_tables_sqlite(sqlite_path: str) -> List[str]:
|
||||
"""Return a list of non-system table names from the SQLite database."""
|
||||
conn = sqlite3.connect(sqlite_path)
|
||||
try:
|
||||
cur = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name;")
|
||||
return [r[0] for r in cur.fetchall()]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def convert_table_sqlite_to_pg(
|
||||
sqlite_path: str,
|
||||
pg_dsn: str,
|
||||
table_name: str,
|
||||
create_table: bool = True,
|
||||
batch_size: int = 1000,
|
||||
) -> None:
|
||||
"""Copy a table from a SQLite file to PostgreSQL.
|
||||
|
||||
Arguments:
|
||||
sqlite_path: path to the .db file
|
||||
pg_dsn: libpq connection string for psycopg2
|
||||
table_name: table to copy
|
||||
create_table: if True, attempt to create the table in Postgres (basic mapping)
|
||||
batch_size: number of rows per insert batch
|
||||
"""
|
||||
LOG.info("Opening SQLite DB %s", sqlite_path)
|
||||
sconn = sqlite3.connect(sqlite_path)
|
||||
sconn.row_factory = sqlite3.Row
|
||||
|
||||
try:
|
||||
info = _get_table_info_sqlite(sconn, table_name)
|
||||
if not info:
|
||||
raise ValueError(f"Table {table_name} not found in {sqlite_path}")
|
||||
|
||||
columns = []
|
||||
pk_cols = []
|
||||
for col in info:
|
||||
# PRAGMA table_info: cid, name, type, notnull, dflt_value, pk
|
||||
name = col[1]
|
||||
ctype = col[2]
|
||||
notnull = bool(col[3])
|
||||
dflt = col[4]
|
||||
pk = bool(col[5])
|
||||
pgtype = _map_sqlite_type_to_pg(ctype)
|
||||
columns.append((name, pgtype, notnull, dflt))
|
||||
if pk:
|
||||
pk_cols.append(name)
|
||||
|
||||
if create_table:
|
||||
LOG.info("Connecting to Postgres to create table %s", table_name)
|
||||
with psycopg2.connect(pg_dsn) as pgconn:
|
||||
with pgconn.cursor() as cur:
|
||||
col_defs = []
|
||||
for name, pgtype, notnull, _dflt in columns:
|
||||
part = sql.SQL("{} {}{}") .format(
|
||||
sql.Identifier(name),
|
||||
sql.SQL(pgtype),
|
||||
sql.SQL(" NOT NULL") if notnull else sql.SQL(""),
|
||||
)
|
||||
col_defs.append(part)
|
||||
|
||||
if pk_cols:
|
||||
pk_clause = sql.SQL(", PRIMARY KEY ({})").format(sql.SQL(",").join(map(sql.Identifier, pk_cols)))
|
||||
create_body = sql.SQL(", ").join(col_defs) + pk_clause
|
||||
else:
|
||||
create_body = sql.SQL(", ").join(col_defs)
|
||||
|
||||
create_stmt = sql.SQL("CREATE TABLE IF NOT EXISTS {} ({})").format(
|
||||
sql.Identifier(table_name), create_body
|
||||
)
|
||||
cur.execute(create_stmt)
|
||||
pgconn.commit()
|
||||
|
||||
# Now copy data
|
||||
LOG.info("Copying data for table %s", table_name)
|
||||
s_cur = sconn.cursor()
|
||||
s_cur.execute(f"SELECT * FROM '{table_name}'")
|
||||
|
||||
cols = [d[0] for d in s_cur.description]
|
||||
placeholders = ",".join(["%s"] * len(cols))
|
||||
insert_sql = sql.SQL("INSERT INTO {} ({}) VALUES ({})").format(
|
||||
sql.Identifier(table_name),
|
||||
sql.SQL(",").join(map(sql.Identifier, cols)),
|
||||
sql.SQL(placeholders),
|
||||
)
|
||||
|
||||
with psycopg2.connect(pg_dsn) as pgconn:
|
||||
with pgconn.cursor() as pgcur:
|
||||
batch = []
|
||||
for row in s_cur:
|
||||
batch.append(tuple(row))
|
||||
if len(batch) >= batch_size:
|
||||
pgcur.executemany(insert_sql.as_string(pgconn), batch)
|
||||
pgconn.commit()
|
||||
batch = []
|
||||
if batch:
|
||||
pgcur.executemany(insert_sql.as_string(pgconn), batch)
|
||||
pgconn.commit()
|
||||
|
||||
finally:
|
||||
sconn.close()
|
||||
|
||||
|
||||
def _cli() -> None:
|
||||
parser = argparse.ArgumentParser(description="Copy one table from SQLite to PostgreSQL")
|
||||
parser.add_argument("sqlite_db", help="Path to SQLite database file")
|
||||
parser.add_argument("pg_dsn", help="Postgres DSN, e.g. \"host=.. dbname=.. user=.. password=..\"")
|
||||
parser.add_argument("table", nargs="?", help="Table name to copy (optional if --tables or --all used)")
|
||||
parser.add_argument("--tables", help="Comma-separated list of tables to copy")
|
||||
parser.add_argument("--all", action="store_true", help="Copy all non-system tables from the SQLite DB")
|
||||
parser.add_argument("--no-create", dest="create", action="store_false", help="Don't attempt to create table in PG")
|
||||
parser.add_argument("--list", dest="list_tables", action="store_true", help="List tables in the SQLite DB and exit")
|
||||
parser.add_argument("--batch", type=int, default=1000, help="Batch size for inserts")
|
||||
parser.add_argument("--verbose", "-v", action="store_true", help="Verbose logging")
|
||||
args = parser.parse_args()
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)
|
||||
if getattr(args, "list_tables", False):
|
||||
for t in list_tables_sqlite(args.sqlite_db):
|
||||
print(t)
|
||||
return
|
||||
|
||||
# Determine tables to process
|
||||
table_names = []
|
||||
if args.all:
|
||||
table_names = list_tables_sqlite(args.sqlite_db)
|
||||
elif args.tables:
|
||||
table_names = [t.strip() for t in args.tables.split(",") if t.strip()]
|
||||
elif args.table:
|
||||
table_names = [args.table]
|
||||
else:
|
||||
parser.error("must specify a table, --tables, or --all unless --list is used")
|
||||
|
||||
if not args.pg_dsn:
|
||||
parser.error("pg_dsn is required to copy tables to Postgres")
|
||||
|
||||
# Process tables sequentially
|
||||
for tbl in table_names:
|
||||
try:
|
||||
LOG.info("Processing table %s", tbl)
|
||||
convert_table_sqlite_to_pg(args.sqlite_db, args.pg_dsn, tbl, create_table=args.create, batch_size=args.batch)
|
||||
except Exception as exc:
|
||||
LOG.exception("Failed to process table %s: %s", tbl, exc)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
_cli()
|
||||
@@ -1,154 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Utility to expose a PostgreSQL table to SQLite by copying it into a sqlite file.
|
||||
|
||||
This is a pragmatic "plugin": it creates a local SQLite table (read-only by default)
|
||||
that mirrors a PostgreSQL table so SQLite queries can run locally against the data.
|
||||
|
||||
Usage examples:
|
||||
# copy single table into sqlite file
|
||||
python sqlite_pg_plugin.py copy --pg "host=... dbname=... user=... password=..." public.mytable /tmp/my.db
|
||||
|
||||
# list tables in Postgres
|
||||
python sqlite_pg_plugin.py list --pg "host=... dbname=... user=... password=..."
|
||||
|
||||
This avoids needing to build a native virtual-table extension. For a direct virtual
|
||||
table (no copy) a C extension using SQLite virtual table API + libpq would be needed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import sqlite3
|
||||
from typing import List, Tuple
|
||||
|
||||
import psycopg2
|
||||
from psycopg2 import sql
|
||||
|
||||
LOG = logging.getLogger("sqlite_pg_plugin")
|
||||
|
||||
|
||||
def list_tables_pg(pg_dsn: str) -> List[str]:
|
||||
q = "SELECT table_schema||'.'||table_name FROM information_schema.tables WHERE table_type='BASE TABLE' AND table_schema NOT IN ('pg_catalog','information_schema') ORDER BY 1;"
|
||||
with psycopg2.connect(pg_dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(q)
|
||||
return [r[0] for r in cur.fetchall()]
|
||||
|
||||
|
||||
def fetch_table_schema_pg(pg_dsn: str, qualified_table: str) -> List[Tuple[str, str, bool]]:
|
||||
"""Return list of (colname, pgtype, notnull) for the given qualified table (schema.table)."""
|
||||
if '.' in qualified_table:
|
||||
schema, table = qualified_table.split('.', 1)
|
||||
else:
|
||||
schema, table = 'public', qualified_table
|
||||
|
||||
q = sql.SQL("SELECT column_name, data_type, is_nullable FROM information_schema.columns WHERE table_schema=%s AND table_name=%s ORDER BY ordinal_position;")
|
||||
with psycopg2.connect(pg_dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(q, (schema, table))
|
||||
out = []
|
||||
for name, dtype, is_nullable in cur.fetchall():
|
||||
out.append((name, dtype, is_nullable == 'NO'))
|
||||
return out
|
||||
|
||||
|
||||
def _map_pg_to_sqlite(pg_type: str) -> str:
|
||||
t = (pg_type or '').lower()
|
||||
if 'int' in t or t in ('bigserial', 'serial'):
|
||||
return 'INTEGER'
|
||||
if any(x in t for x in ('char', 'text', 'uuid')):
|
||||
return 'TEXT'
|
||||
if any(x in t for x in ('timestamp', 'date', 'time')):
|
||||
return 'TEXT'
|
||||
if any(x in t for x in ('numeric', 'decimal', 'real', 'double', 'float')):
|
||||
return 'REAL'
|
||||
if 'bool' in t:
|
||||
return 'INTEGER'
|
||||
return 'TEXT'
|
||||
|
||||
|
||||
def copy_table_pg_to_sqlite(pg_dsn: str, qualified_table: str, sqlite_path: str, sqlite_table: str | None = None, batch_size: int = 1000, overwrite: bool = True) -> None:
|
||||
"""Copy schema and data from Postgres table into a SQLite file.
|
||||
|
||||
- `qualified_table` can be 'schema.table' or 'table'
|
||||
- `sqlite_table` is optional name to use in sqlite; defaults to table part
|
||||
"""
|
||||
if '.' in qualified_table and not sqlite_table:
|
||||
sqlite_table = qualified_table.split('.', 1)[1]
|
||||
elif not sqlite_table:
|
||||
sqlite_table = qualified_table
|
||||
|
||||
LOG.info('Fetching schema for %s', qualified_table)
|
||||
cols = fetch_table_schema_pg(pg_dsn, qualified_table)
|
||||
if not cols:
|
||||
raise ValueError(f'No columns found for {qualified_table}')
|
||||
|
||||
# create sqlite table
|
||||
sconn = sqlite3.connect(sqlite_path)
|
||||
try:
|
||||
s_cur = sconn.cursor()
|
||||
if overwrite:
|
||||
s_cur.execute(f"DROP TABLE IF EXISTS '{sqlite_table}'")
|
||||
|
||||
col_defs = []
|
||||
for name, pgtype, notnull in cols:
|
||||
stype = _map_pg_to_sqlite(pgtype)
|
||||
col_defs.append(f"{name} {stype}{' NOT NULL' if notnull else ''}")
|
||||
|
||||
create_sql = f"CREATE TABLE IF NOT EXISTS '{sqlite_table}' ({', '.join(col_defs)})"
|
||||
s_cur.execute(create_sql)
|
||||
sconn.commit()
|
||||
|
||||
# stream rows from Postgres
|
||||
with psycopg2.connect(pg_dsn) as pgconn:
|
||||
with pgconn.cursor(name='stream_cursor') as pgcur:
|
||||
pgcur.itersize = batch_size
|
||||
pgcur.execute(sql.SQL('SELECT {} FROM {}').format(sql.SQL(',').join(sql.Identifier(c[0]) for c in cols), sql.Identifier(*qualified_table.split('.', 1)) if '.' in qualified_table else sql.Identifier(qualified_table)))
|
||||
|
||||
placeholders = ','.join('?' for _ in cols)
|
||||
insert_sql = f"INSERT INTO '{sqlite_table}' ({', '.join(n for n,_,_ in cols)}) VALUES ({placeholders})"
|
||||
batch = []
|
||||
for row in pgcur:
|
||||
batch.append(tuple(row))
|
||||
if len(batch) >= batch_size:
|
||||
s_cur.executemany(insert_sql, batch)
|
||||
sconn.commit()
|
||||
batch = []
|
||||
if batch:
|
||||
s_cur.executemany(insert_sql, batch)
|
||||
sconn.commit()
|
||||
finally:
|
||||
sconn.close()
|
||||
|
||||
|
||||
def _cli() -> None:
|
||||
parser = argparse.ArgumentParser(description='Expose Postgres table(s) to SQLite by copying them')
|
||||
sub = parser.add_subparsers(dest='cmd', required=True)
|
||||
|
||||
p_list = sub.add_parser('list', help='List tables in Postgres')
|
||||
p_list.add_argument('--pg', required=True, help='Postgres DSN')
|
||||
|
||||
p_copy = sub.add_parser('copy', help='Copy a Postgres table into sqlite file')
|
||||
p_copy.add_argument('--pg', required=True, help='Postgres DSN')
|
||||
p_copy.add_argument('table', help='Qualified table name (schema.table or table)')
|
||||
p_copy.add_argument('sqlite', help='Path to sqlite file to write to')
|
||||
p_copy.add_argument('--target', help='Target sqlite table name (defaults to original table)')
|
||||
p_copy.add_argument('--batch', type=int, default=1000)
|
||||
p_copy.add_argument('--no-overwrite', dest='overwrite', action='store_false')
|
||||
p_copy.add_argument('--verbose', '-v', action='store_true')
|
||||
|
||||
args = parser.parse_args()
|
||||
logging.basicConfig(level=logging.DEBUG if getattr(args, 'verbose', False) else logging.INFO)
|
||||
|
||||
if args.cmd == 'list':
|
||||
for t in list_tables_pg(args.pg):
|
||||
print(t)
|
||||
return
|
||||
|
||||
if args.cmd == 'copy':
|
||||
copy_table_pg_to_sqlite(args.pg, args.table, args.sqlite, sqlite_table=args.target, batch_size=args.batch, overwrite=args.overwrite)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
_cli()
|
||||
@@ -82,9 +82,9 @@ def main() -> int:
|
||||
directories["CASTDIR"] = os.path.join(ROOTDIR, "cache", "cast") + os.sep
|
||||
directories["GUESTCASTDIR"] = os.path.join(ROOTDIR, "cache", "guestcast") + os.sep
|
||||
directories["CREWDIR"] = os.path.join(ROOTDIR, "cache", "crew") + os.sep
|
||||
directories["GUESTCREWDIR"] = os.path.join(ROOTDIR, "cache", "guestcrew") + os.sep
|
||||
directories["CREDITSDIR"] = os.path.join(ROOTDIR, "cache", "credits") + os.sep
|
||||
directories["ALIASDIR"] = os.path.join(ROOTDIR, "cache", "aliases") + os.sep
|
||||
#directories["GUESTCREWDIR"] = os.path.join(ROOTDIR, "cache", "guestcrew") + os.sep
|
||||
#directories["CREDITSDIR"] = os.path.join(ROOTDIR, "cache", "credits") + os.sep
|
||||
#directories["ALIASDIR"] = os.path.join(ROOTDIR, "cache", "aliases") + os.sep
|
||||
tempdir = os.path.join(ROOTDIR, 'temp') + os.sep
|
||||
updatetable = 'updates.tvupdates'
|
||||
skip_file = f"{tempdir}skip_ids.txt"
|
||||
@@ -297,7 +297,7 @@ def main() -> int:
|
||||
try:
|
||||
if int(ts) == int(prev_ts):
|
||||
found_path = find_file_in_multiple_dirs(f"{seriesid}.json", directory_list)
|
||||
if len(found_path) == 8:
|
||||
if len(found_path) == 5:
|
||||
continue
|
||||
else:
|
||||
for directory in directory_list:
|
||||
Reference in New Issue
Block a user