wjones/new_dbsync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'Refactor README.md for clarity and detail: enhance project description, add features, usage examples, and troubleshooting sections; remove outdated mongo2pgsql.py and search_mongodb.py files' (#6) from development into main

Browse Source 
Reviewed-on: #6
This commit is contained in:
Wendell Jones
2026-02-06 18:05:01 +00:00
parent a80c692786 1283bd282a
commit 4f7c88e96e
3 changed files with 159 additions and 476 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+159 -5
View File
@@ -1,8 +1,162 @@
# DBSyncer
# DBSyncer - MongoDB to PostgreSQL Data Synchronization
This project has two parts:
1.) it uses the tvmaze API to download Television dat and store it in a mongodb database.
2.) Takes saved json data and inserts/updates data in a postgresql database.
A comprehensive data synchronization tool that:
1. Downloads Television data from the TVMaze API and stores it in MongoDB
2. Automatically generates PostgreSQL table schemas from MongoDB document structures
3. Efficiently inserts/updates MongoDB documents into PostgreSQL with intelligent type conversion
## Features
[Learn more about creating GitLab projects.](https://docs.gitlab.com/ee/gitlab-basics/create-project.html)
### Schema Generation (`db/schema_generator.py`)
- **Automatic Schema Inference**: Analyzes MongoDB documents to infer PostgreSQL types
- **Type Mapping**: Intelligent conversion between MongoDB and PostgreSQL types:
- Objects/Arrays → JSONB
- Dates → TIMESTAMP
- Numbers → BIGINT / DOUBLE PRECISION
- Strings → TEXT
### Document Insertion
- **Batch Processing**: Efficiently inserts documents in configurable batches (default: 1000)
- **Type Conversion**: Automatic type conversion for:
- Epoch timestamps ↔ ISO datetime strings
- JSON strings ↔ Python dict/list
- ObjectId → Text
- **UPSERT Support**: Uses PostgreSQL `ON CONFLICT ... DO UPDATE SET` for automatic upserts
- **Error Handling**: Detailed logging with per-row error reporting
## Quick Start
### Prerequisites
```bash
pip install -r requirements.txt
```
### Configuration
Edit `settings/tvsync_settings.cfg`:
```ini
[dbsettings]
hostname = localhost
pgsqlport = 5432
pgsqlusername = postgres
pgsqlpassword = your_password
dbname = media_dbsync
mghostname = localhost
mgport = 27017
mgdbname = media
[database]
dbtype = pgsql
updateschema = updates
```
### Main Scripts
#### `update_mongodb.py`
Fetches TV data from TVMaze API and stores in MongoDB:
```bash
python3 update_mongodb.py
```
#### `mongodb2postgres.py`
Generates PostgreSQL schemas from MongoDB collections and inserts documents:
```bash
python3 mongodb2postgres.py
```
This script will:
1. Analyze each MongoDB collection (series, episodes, actors, characters, crew)
2. Automatically create corresponding PostgreSQL tables in the `updates` schema
3. Insert/upsert all documents with proper type conversions
4. Report progress and any errors
## Usage Examples
### Generate Table Schema from MongoDB Collection
```python
from db.schema_generator import MongoToPostgresSchemaGenerator
generator = MongoToPostgresSchemaGenerator(sample_size=100)
schema = generator.analyze_collection(mongo_db, 'series')
# View the SQL
sql = generator.generate_create_table_sql('seriesdata', schema_name='dbo', pk_field='id')
print(sql)
# Create the table
generator.create_table_in_postgres(engine, 'seriesdata', schema_name='dbo', pk_field='id')
```
### Insert Documents with Type Conversion
```python
from db.schema_generator import MongoDocumentInserter
inserter = MongoDocumentInserter(batch_size=1000)
# From a collection
count = inserter.insert_from_collection(
engine=engine,
table_name='seriesdata',
collection=mongo_db['series'],
schema_name='dbo',
on_conflict="ON CONFLICT (id) DO UPDATE SET name=EXCLUDED.name, updated=EXCLUDED.updated",
)
print(f"Inserted {count} documents")
```
## Architecture
### Database Classes (`db/functions.py`)
- `settype`: Manages database engine creation and API configuration
- `dbmongo`: Handles MongoDB connections and operations
- `Dbexec`: Executes raw SQL queries and batch updates
### Schema Generator (`db/schema_generator.py`)
- `MongoToPostgresSchemaGenerator`: Infers and creates table schemas
- `MongoDocumentInserter`: Handles document insertion with type conversion
### API Integration (`api/`)
- TVMaze API client for fetching show, episode, cast, and crew data
- TheTVDB API client (optional)
## Performance Optimizations
- **Batch Inserts**: Uses SQLAlchemy's parameterized queries for efficient batch operations
- **Transactional Inserts**: Groups inserts in transactions to reduce database overhead
- **Column Type Awareness**: Converts data to match target column types, avoiding casting overhead
- **Connection Pooling**: Reuses database connections (pool_size=20, max_overflow=20)
- **UPSERT Operations**: Uses PostgreSQL's `ON CONFLICT` for atomic insert-or-update
## Troubleshooting
### MetaData Schema Argument Error
If you see `sqlalchemy.exc.ArgumentError: Could not parse SQLAlchemy URL`, ensure:
- Use `MetaData(schema="updates")` instead of `MetaData("updates")`
- Pass full connection strings to `create_engine()`, not schema names
### Type Conversion Errors
The inserter automatically handles:
- Datetime strings → epoch timestamps (for BIGINT columns)
- Epoch ints → datetime objects (for TIMESTAMP columns)
- JSON strings → Python dicts (for JSONB columns)
If type errors persist, check that:
1. Table schema matches MongoDB document fields
2. Column types are correctly inferred from sample documents
3. Use appropriate `on_conflict` clauses for upserts
### Column Name Case Sensitivity
PostgreSQL stores unquoted identifiers as lowercase. The inserter automatically:
- Lowercases column names in `ON CONFLICT` clauses
- Uses case-insensitive matching when mapping MongoDB fields to PostgreSQL columns
## Contributing
When modifying:
- **Schema inference**: Update `MongoToPostgresSchemaGenerator.TYPE_MAPPING` for new type support
- **Type conversion**: Extend `MongoDocumentInserter.convert_value()` for custom conversions
- **Database operations**: Use parameterized queries via `text()` to prevent SQL injection
- **Error handling**: Add specific exception types to improve debugging
## License
See LICENSE file for details.
mongo2pgsql.py
-205
View File
@@ -1,205 +0,0 @@
#!/usr/bin/env python3
"""Convert a MongoDB collection into a PostgreSQL table.
Contains `convert_mongo_to_pg(mongo_uri, mongo_db, collection_name, pg_dsn, pg_table=None, ...)`
and a small CLI.
Notes:
- Uses `pymongo` to read documents and `psycopg2` to write to Postgres.
- Infers basic column types from a sample of documents; nested objects/arrays
are stored as `JSONB`.
"""
from __future__ import annotations
import argparse
import json
import logging
from typing import Any, Dict, Iterable, List, Optional
import pymongo
import psycopg2
import psycopg2.extras as pgextras
import psycopg2.sql as sql
LOG = logging.getLogger("mongo2pgsql")
def _detect_type(value: Any) -> str:
if value is None:
return "null"
if isinstance(value, bool):
return "boolean"
if isinstance(value, int) and not isinstance(value, bool):
return "integer"
if isinstance(value, float):
return "real"
if isinstance(value, (dict, list)):
return "jsonb"
return "text"
def _merge_types(types: List[str]) -> str:
# priority: jsonb > text > real > integer > boolean > null
s = set(types)
if "jsonb" in s:
return "jsonb"
if "text" in s:
return "text"
if "real" in s:
return "real"
if "integer" in s:
return "integer"
if "boolean" in s:
return "boolean"
return "text"
def _pg_type_for(kind: str) -> str:
return {
"integer": "BIGINT",
"real": "DOUBLE PRECISION",
"boolean": "BOOLEAN",
"jsonb": "JSONB",
"text": "TEXT",
}.get(kind, "TEXT")
def convert_mongo_to_pg(
mongo_uri: str,
mongo_db: str,
collection_name: str,
pg_dsn: str,
pg_table: Optional[str] = None,
sample_size: int = 500,
batch_size: int = 1000,
create_table: bool = True,
pk_field: str = "_id",
) -> None:
"""Copy a MongoDB collection to PostgreSQL.
- `pg_table` defaults to `collection_name` if omitted.
- Nested documents/arrays are stored as JSONB.
- `_id` is stored as TEXT by default.
"""
if pg_table is None:
pg_table = collection_name
client = pymongo.MongoClient(mongo_uri)
coll = client[mongo_db][collection_name]
sample = list(coll.find({}, projection=None, limit=sample_size))
if not sample:
raise ValueError("collection is empty or not found")
# gather all field names and observed types
field_types: Dict[str, List[str]] = {}
for doc in sample:
for k, v in doc.items():
t = _detect_type(v)
field_types.setdefault(k, []).append(t)
# finalize types
final_types: Dict[str, str] = {}
for k, types in field_types.items():
merged = _merge_types(types)
final_types[k] = _pg_type_for(merged)
# ensure primary key present
if pk_field not in final_types:
final_types[pk_field] = "TEXT"
cols = list(final_types.items())
# create table
if create_table:
# Build CREATE TABLE statement using psycopg2.sql for safe identifiers
col_defs = []
for name, ptype in cols:
col_defs.append(sql.SQL("{} {}").format(sql.Identifier(name), sql.SQL(ptype)))
create_body = sql.SQL(", ").join(col_defs)
if pk_field in final_types:
pk_clause = sql.SQL(", PRIMARY KEY ({})").format(sql.SQL(",").join(map(sql.Identifier, [pk_field])))
create_body = create_body + pk_clause
create_stmt = sql.SQL("CREATE TABLE IF NOT EXISTS {} ({})").format(sql.Identifier(pg_table), create_body)
with psycopg2.connect(pg_dsn) as pgconn:
with pgconn.cursor() as cur:
LOG.info("Creating table %s", pg_table)
cur.execute(create_stmt)
pgconn.commit()
# stream documents and insert
insert_cols = [n for n, _ in cols]
with psycopg2.connect(pg_dsn) as pgconn:
with pgconn.cursor() as cur:
# build INSERT statement for execute_values: it expects a single %s
base_insert = sql.SQL("INSERT INTO {} ({}) VALUES %s").format(
sql.Identifier(pg_table),
sql.SQL(', ').join(map(sql.Identifier, insert_cols)),
).as_string(pgconn)
batch = []
# Use an explicit session when using no_cursor_timeout to avoid pymongo warning
with client.start_session() as session:
cursor = coll.find({}, no_cursor_timeout=True, session=session).batch_size(batch_size)
try:
for doc in cursor:
row = []
for col_name in insert_cols:
v = doc.get(col_name)
if v is None:
row.append(None)
continue
# convert ObjectId and other non-JSON types
if isinstance(v, (dict, list)):
row.append(pgextras.Json(v))
else:
row.append(str(v) if col_name == pk_field else v)
batch.append(tuple(row))
if len(batch) >= batch_size:
pgextras.execute_values(cur, base_insert, batch, template=None, page_size=batch_size)
pgconn.commit()
batch = []
finally:
try:
cursor.close()
except Exception:
pass
if batch:
pgextras.execute_values(cur, base_insert, batch, template=None, page_size=len(batch))
pgconn.commit()
LOG.info("Finished copying collection %s.%s to %s", mongo_db, collection_name, pg_table)
def _cli() -> None:
parser = argparse.ArgumentParser(description="Copy a MongoDB collection to PostgreSQL")
parser.add_argument("mongo_uri", help="MongoDB connection URI, e.g. mongodb://user:pw@host:27017")
parser.add_argument("mongo_db", help="MongoDB database name")
parser.add_argument("collection", help="Collection name to copy")
parser.add_argument("pg_dsn", help="Postgres DSN, e.g. \"host=.. dbname=.. user=.. password=..\"")
parser.add_argument("--pg-table", help="Target Postgres table name (defaults to collection name)")
parser.add_argument("--sample", type=int, default=500, help="Sample size to infer schema")
parser.add_argument("--batch", type=int, default=1000, help="Insert batch size")
parser.add_argument("--no-create", dest="create", action="store_false", help="Don't attempt to create table in PG")
parser.add_argument("--verbose", "-v", action="store_true")
args = parser.parse_args()
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)
convert_mongo_to_pg(
args.mongo_uri,
args.mongo_db,
args.collection,
args.pg_dsn,
pg_table=args.pg_table,
sample_size=args.sample,
batch_size=args.batch,
create_table=args.create,
)
if __name__ == "__main__":
_cli()
search mongodb.py
-266
View File
@@ -1,266 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os, time, sys, datetime
from db.functions import settype, dbmongo
import json
def set_apienv(urls, uprocess, dbengine, dbExec, updatesBase, lprint):
"""Populate updates table from the API and return available updates.
Args:
urls: URL configuration with an "updatesurl" key.
uprocess: URL loader instance with `get_data`.
jget: JSON processor with `jconvert`.
dbengine: database engine dict (expects key "engine").
dbExec: database execution helper with `update_tvupdates` and `rawsql_select`.
updatesBase: list-like containing tables (uses index 8 for updates table).
lprint: logger instance with `logprint`.
Returns:
List of tuples (seriesid, timestamp) representing available updates.
"""
#updatetable = updatesBase[8]
#inputdata = uprocess.get_data(urls["updatesurl"])
#lprint.logprint("info", f"Retrieved {len(availableupdates)} rows for processing......")
#
newupdates = dbExec.rawsql_select(
dbengine["engine"],
"select seriesid,timestamp from updates.tvupdates",
lprint
)
return newupdates
def print_starttime(lprint, ts1):
lprint.logprint("info", f"Sync start time: {str(ts1)}")
def find_file_in_multiple_dirs(filename, directories):
"""
Checks if a file exists in any of the provided directories.
Args:
filename (str): The name of the file to search for.
directories (list): A list of directory paths to check.
Returns:
Path or None: The full path to the file if found, otherwise None.
"""
directories_found = []
for directory in directories:
if os.path.isfile(f'{directory}{filename}'):
directories_found.append(directory)
return directories_found
def load_json_to_mongodb(mongo_updater, directories, update_list, updateDict, lprint, track_changes=True):
"""
Load downloaded JSON files into MongoDB collections with change detection.
Processes series, episodes, cast, and crew data from cached JSON files
and inserts or updates them in the appropriate MongoDB collections.
Automatically detects if documents are new, updated, or unchanged by comparing
updateDict timestamps with existing document 'updated' fields.
Args:
mongo_updater: dbmongo instance for MongoDB operations
directories: dict containing cache directory paths for different data types
update_list: list of series IDs to process
updateDict: dict with seriesid as key and timestamp as value
lprint: logger instance for logging operations
track_changes: bool, if True tracks inserted vs updated documents (default: True)
Returns:
tuple: (successful_count, failed_count, stats_dict) for series processed
stats_dict contains: {'inserted': count, 'updated': count}
"""
successful_count = 0
failed_count = 0
stats = {'series_inserted': 0, 'series_updated': 0, 'series_skipped': 0,
'episodes_inserted': 0, 'episodes_updated': 0, 'episodes_skipped': 0,
'cast_inserted': 0, 'cast_updated': 0, 'cast_skipped': 0,
'crew_inserted': 0, 'crew_updated': 0, 'crew_skipped': 0}
lprint.logprint("info", f"Starting to load {len(update_list)} series into MongoDB")
for seriesid in tqdm(update_list, desc="Loading series data into MongoDB", unit="series"):
try:
# Load series data
series_file = f"{directories['SERIESDIR']}{seriesid}.json"
if os.path.exists(series_file):
try:
with open(series_file, 'r') as f:
series_data = json.load(f)
# Insert or update series in MongoDB
if series_data:
if track_changes:
existing = mongo_updater.mgseries.find_one({"id": series_data.get("id")})
seriesid_str = str(seriesid)
update_timestamp = updateDict.get(seriesid_str)
if existing:
# Check if data has changed by comparing update timestamp
if update_timestamp and int(update_timestamp) > existing.get("updated", 0):
mongo_updater.bulk_upsert_by_id(mongo_updater.mgseries, [series_data])
stats['series_updated'] += 1
lprint.logprint("debug", f"Updated series data for ID {seriesid}")
else:
lprint.logprint("debug", f"Series {seriesid} unchanged, skipping")
stats['series_skipped'] += 1
pass # Document unchanged
else:
mongo_updater.bulk_upsert_by_id(mongo_updater.mgseries, [series_data])
stats['series_inserted'] += 1
lprint.logprint("debug", f"Inserted new series data for ID {seriesid}")
# Load episodes data
episodes_file = f"{directories['EPISODEDIR']}{seriesid}.json"
if os.path.exists(episodes_file):
try:
with open(episodes_file, 'r') as f:
episodes_data = json.load(f)
if isinstance(episodes_data, list) and episodes_data:
# Add seriesid to each episode
for episode in episodes_data:
episode['seriesid'] = seriesid
if track_changes:
for episode in episodes_data:
existing = mongo_updater.mgepisodes.find_one({"id": episode.get("id")})
seriesid_str = str(seriesid)
update_timestamp = updateDict.get(seriesid_str)
if existing:
if update_timestamp and int(update_timestamp) > existing.get("updated", 0):
mongo_updater.bulk_upsert_by_id(mongo_updater.mgepisodes, episodes_data)
stats['episodes_updated'] += 1
else:
lprint.logprint("debug", f"Episode Data for Series {seriesid} unchanged, skipping")
pass # Document unchanged
else:
mongo_updater.bulk_upsert_by_id(mongo_updater.mgepisodes, episodes_data)
stats['episodes_inserted'] += 1
#mongo_updater.bulk_upsert_by_id(mongo_updater.mgepisodes, episodes_data)
lprint.logprint("debug", f"Loaded {len(episodes_data)} episodes for series {seriesid}")
except Exception as e:
lprint.logprint("warning", f"Failed to load episodes file for {seriesid}: {e}")
# Load cast data
cast_file = f"{directories['CASTDIR']}{seriesid}.json"
if os.path.exists(cast_file):
try:
with open(cast_file, 'r') as f:
cast_data = json.load(f)
if isinstance(cast_data, list) and cast_data:
actors = []
characters = []
for item in cast_data:
if 'person' in item and item['person']:
item['person']['seriesid'] = seriesid
actors.append(item['person'])
if 'character' in item and item['character']:
item['character']['seriesid'] = seriesid
characters.append(item['character'])
if track_changes:
for actor in actors:
existing = mongo_updater.mgactors.find_one({"id": actor.get("id")})
seriesid_str = str(seriesid)
update_timestamp = updateDict.get(seriesid_str)
if existing:
if update_timestamp and int(update_timestamp) > existing.get("updated", 0):
stats['cast_updated'] += 1
else:
stats['cast_inserted'] += 1
if actors:
mongo_updater.bulk_upsert_by_id(mongo_updater.mgactors, actors)
if characters:
mongo_updater.bulk_upsert_by_id(mongo_updater.mgcharacters, characters)
lprint.logprint("debug", f"Loaded cast data for series {seriesid}")
except Exception as e:
lprint.logprint("warning", f"Failed to load cast file for {seriesid}: {e}")
# Load crew data
crew_file = f"{directories['CREWDIR']}{seriesid}.json"
if os.path.exists(crew_file):
try:
with open(crew_file, 'r') as f:
crew_data = json.load(f)
if isinstance(crew_data, list) and crew_data:
crew_list = []
for item in crew_data:
if 'person' in item and item['person']:
item['person']['seriesid'] = seriesid
item['person']['crew_type'] = item.get('type', 'unknown')
crew_list.append(item['person'])
if track_changes:
for crew in crew_list:
existing = mongo_updater.mgcrew.find_one({"id": crew.get("id")})
seriesid_str = str(seriesid)
update_timestamp = updateDict.get(seriesid_str)
if existing:
if update_timestamp and int(update_timestamp) > existing.get("updated", 0):
stats['crew_updated'] += 1
else:
stats['crew_inserted'] += 1
if crew_list:
mongo_updater.bulk_upsert_by_id(mongo_updater.mgcrew, crew_list)
lprint.logprint("debug", f"Loaded crew data for series {seriesid}")
except Exception as e:
lprint.logprint("warning", f"Failed to load crew file for {seriesid}: {e}")
successful_count += 1
lprint.logprint("info", f"Successfully processed series {seriesid}")
#mongo_updater.bulk_upsert_by_id(mongo_updater.mgseries, [series_data])
except Exception as e:
lprint.logprint("warning", f"Failed to load series file for {seriesid}: {e}")
failed_count += 1
continue
except Exception as e:
failed_count += 1
lprint.logprint("error", f"Unexpected error processing series {seriesid}: {e}")
lprint.logprint("info", f"MongoDB loading complete.")
return successful_count, failed_count, stats
def main() -> int:
ROOTDIR = os.getcwd()
directories = {}
# Use cross-platform path builders and ensure directories exist
LOGDIR = os.path.join(ROOTDIR, "log")
directories["SERIESDIR"] = os.path.join(ROOTDIR, "cache", "series") + os.sep
directories["EPISODEDIR"] = os.path.join(ROOTDIR, "cache", "episodes") + os.sep
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
tempdir = os.path.join(ROOTDIR, 'temp') + os.sep
import settings.config
updateList = []
updateDict = {}
needed_updates = []
config_options = settings.config.Config(ROOTDIR)
options = config_options.config_options
mongo_query = dbmongo(options)
seriesid = 4
query = { "id": int(seriesid)}
for doc in mongo_query.mgseries.find(query):
print(doc)
query = { "seriesid": str(seriesid)}
for doc in mongo_query.mgepisodes.find(query):
print(doc)
if __name__ == "__main__":
sys.exit(main())