Last active
January 7, 2025 22:02
-
-
Save brandonrobertz/e054a8bab22e2ea41b655bfe5fec1ca9 to your computer and use it in GitHub Desktop.
PostgreSQL to SQLite3 Database
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python | |
| import os | |
| #!pip install psycopg2-binary | |
| import psycopg2 | |
| import psycopg2.extras | |
| #!pip install sqlite-utils | |
| import sqlite_utils | |
| DATABASE=os.getenv("DB_NAME") or os.getenv("POSTGRES_DB") | |
| USER=os.getenv("DB_USER") or os.getenv("POSTGRES_USER") | |
| HOST=os.getenv("DB_HOST") or os.getenv("POSTGRES_HOST") | |
| PASSWORD=os.getenv("DB_PASS") or os.getenv("POSTGRES_PASSWORD") | |
| # comma separated for multiple (no spaces) | |
| SCHEMAS=os.getenv("DB_SCHEMAS") or os.getenv("POSTGRES_SCHEMAS", "public") | |
| OUTPUT_SQLITE_DB="output.db" | |
| # List of tables to dump | |
| ONLY_TABLES = None | |
| assert DATABASE and HOST, "Make sure to fill in the postgres DB connection variables!" | |
| print(f"Converting psql DB {USER}@{HOST}:{DATABASE} to {OUTPUT_SQLITE_DB}...") | |
| db = sqlite_utils.Database(OUTPUT_SQLITE_DB) | |
| db.enable_wal() | |
| conn = psycopg2.connect(database=DATABASE, user=USER, host=HOST, password=PASSWORD) | |
| conn = psycopg2.connect(database=DATABASE, user=USER, host=HOST, password=PASSWORD, options=f"-c search_path={SCHEMAS}") | |
| cursor = conn.cursor() | |
| cursor.execute("select relname from pg_class where relkind='r' and relname !~ '^(pg_|sql_)';") | |
| for (table,) in cursor.fetchall(): | |
| print(f"Loading table {table}") | |
| if ONLY_TABLES and table not in ONLY_TABLES: | |
| print(f"Skipping non-included table {table}") | |
| continue | |
| table_cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor) | |
| data = [] | |
| table_cur.execute("SELECT * FROM %s" % (table)) | |
| for i, row in enumerate(table_cur.fetchall()): | |
| # data.append(dict(row)) | |
| if i % 1000 == 0: | |
| print(table, i, end="\r") | |
| db[table].insert(dict(row)) | |
| print(f"Completed extracting {table}") | |
| conn.close() | |
| db.disable_wal() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment