Add MongoDB query utilities and update batch size for document insertion
This commit is contained in:
@@ -0,0 +1,270 @@
|
||||
"""MongoDB query utilities for filtering documents by update timestamp."""
|
||||
|
||||
__author__ = 'Wendell Jones'
|
||||
|
||||
from typing import Any, Dict, List, Optional
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MongoQueryUtils:
|
||||
"""Utilities for querying MongoDB collections."""
|
||||
|
||||
@staticmethod
|
||||
def get_recent_updates(
|
||||
collection: Any,
|
||||
minutes: int = 60,
|
||||
updated_field: str = 'updated',
|
||||
query_filter: Optional[Dict] = None,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Find documents updated in the last N minutes.
|
||||
|
||||
Handles both epoch timestamps (int/float) and ISO datetime strings.
|
||||
|
||||
Args:
|
||||
collection: PyMongo collection object.
|
||||
minutes: Number of minutes to look back (default: 60).
|
||||
updated_field: Name of the updated timestamp field (default: 'updated').
|
||||
query_filter: Additional MongoDB query filter to apply (optional).
|
||||
|
||||
Returns:
|
||||
List of documents updated within the time window.
|
||||
|
||||
Example:
|
||||
from db.query_utils import MongoQueryUtils
|
||||
|
||||
recent = MongoQueryUtils.get_recent_updates(
|
||||
collection=mongo_db['series'],
|
||||
minutes=60,
|
||||
updated_field='updated',
|
||||
)
|
||||
print(f"Found {len(recent)} recently updated documents")
|
||||
"""
|
||||
if query_filter is None:
|
||||
query_filter = {}
|
||||
|
||||
# Calculate cutoff time (60 minutes ago)
|
||||
now = datetime.now(timezone.utc)
|
||||
cutoff_time = now - timedelta(minutes=minutes)
|
||||
cutoff_epoch = int(cutoff_time.timestamp())
|
||||
|
||||
logger.info(
|
||||
f"Searching for documents updated in last {minutes} minutes "
|
||||
f"(since {cutoff_time.isoformat()})"
|
||||
)
|
||||
|
||||
# Build query for documents updated after cutoff
|
||||
# Support both epoch timestamps and ISO datetime strings
|
||||
query = {
|
||||
**query_filter,
|
||||
'$or': [
|
||||
# Epoch timestamp (int/float)
|
||||
{updated_field: {'$gte': cutoff_epoch}},
|
||||
# ISO datetime string
|
||||
{updated_field: {'$gte': cutoff_time.isoformat()}},
|
||||
]
|
||||
}
|
||||
|
||||
documents = list(collection.find(query))
|
||||
logger.info(f"Found {len(documents)} documents matching criteria")
|
||||
|
||||
return documents
|
||||
|
||||
@staticmethod
|
||||
def get_recent_updates_by_id(
|
||||
collection: Any,
|
||||
minutes: int = 60,
|
||||
updated_field: str = 'updated',
|
||||
id_field: str = 'id',
|
||||
query_filter: Optional[Dict] = None,
|
||||
) -> List[int]:
|
||||
"""
|
||||
Get IDs of documents updated in the last N minutes.
|
||||
|
||||
Useful for batch processing or sync operations.
|
||||
|
||||
Args:
|
||||
collection: PyMongo collection object.
|
||||
minutes: Number of minutes to look back (default: 60).
|
||||
updated_field: Name of the updated timestamp field (default: 'updated').
|
||||
id_field: Name of the ID field (default: 'id').
|
||||
query_filter: Additional MongoDB query filter (optional).
|
||||
|
||||
Returns:
|
||||
List of document IDs that were recently updated.
|
||||
|
||||
Example:
|
||||
ids = MongoQueryUtils.get_recent_updates_by_id(
|
||||
collection=mongo_db['series'],
|
||||
minutes=60,
|
||||
)
|
||||
print(f"Series IDs to update: {ids}")
|
||||
"""
|
||||
docs = MongoQueryUtils.get_recent_updates(
|
||||
collection=collection,
|
||||
minutes=minutes,
|
||||
updated_field=updated_field,
|
||||
query_filter=query_filter,
|
||||
)
|
||||
|
||||
ids = [doc.get(id_field) for doc in docs if id_field in doc]
|
||||
return ids
|
||||
|
||||
@staticmethod
|
||||
def get_updates_since(
|
||||
collection: Any,
|
||||
since_timestamp: Optional[Any] = None,
|
||||
minutes: int = 60,
|
||||
updated_field: str = 'updated',
|
||||
query_filter: Optional[Dict] = None,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Find documents updated since a specific timestamp or N minutes ago.
|
||||
|
||||
Args:
|
||||
collection: PyMongo collection object.
|
||||
since_timestamp: Epoch int, float, or ISO datetime string (optional).
|
||||
If None, defaults to N minutes ago (see minutes parameter).
|
||||
minutes: If since_timestamp is None, look back this many minutes (default: 60).
|
||||
updated_field: Name of the updated timestamp field (default: 'updated').
|
||||
query_filter: Additional MongoDB query filter (optional).
|
||||
|
||||
Returns:
|
||||
List of documents updated since the timestamp.
|
||||
|
||||
Example:
|
||||
# Get updates since a specific timestamp
|
||||
cutoff = 1704067200 # Epoch timestamp
|
||||
docs = MongoQueryUtils.get_updates_since(
|
||||
collection=mongo_db['series'],
|
||||
since_timestamp=cutoff,
|
||||
)
|
||||
|
||||
# Or use default 60 minutes ago
|
||||
docs = MongoQueryUtils.get_updates_since(
|
||||
collection=mongo_db['series'],
|
||||
)
|
||||
|
||||
# Or specify different time window
|
||||
docs = MongoQueryUtils.get_updates_since(
|
||||
collection=mongo_db['series'],
|
||||
minutes=120, # Last 2 hours
|
||||
)
|
||||
"""
|
||||
if query_filter is None:
|
||||
query_filter = {}
|
||||
|
||||
# If no timestamp provided, use minutes parameter to calculate cutoff
|
||||
if since_timestamp is None:
|
||||
now = datetime.now(timezone.utc)
|
||||
cutoff_time = now - timedelta(minutes=minutes)
|
||||
since_timestamp = int(cutoff_time.timestamp())
|
||||
logger.info(f"Using default {minutes} minutes ago: {cutoff_time.isoformat()}")
|
||||
|
||||
query = {
|
||||
**query_filter,
|
||||
'$or': [
|
||||
{updated_field: {'$gte': since_timestamp}},
|
||||
{updated_field: {'$gte': str(since_timestamp)}},
|
||||
]
|
||||
}
|
||||
|
||||
documents = list(collection.find(query))
|
||||
logger.info(f"Found {len(documents)} documents updated since {since_timestamp}")
|
||||
|
||||
return documents
|
||||
|
||||
@staticmethod
|
||||
def count_recent_updates(
|
||||
collection: Any,
|
||||
minutes: int = 60,
|
||||
updated_field: str = 'updated',
|
||||
query_filter: Optional[Dict] = None,
|
||||
) -> int:
|
||||
"""
|
||||
Count documents updated in the last N minutes.
|
||||
|
||||
Args:
|
||||
collection: PyMongo collection object.
|
||||
minutes: Number of minutes to look back (default: 60).
|
||||
updated_field: Name of the updated timestamp field (default: 'updated').
|
||||
query_filter: Additional MongoDB query filter (optional).
|
||||
|
||||
Returns:
|
||||
Count of recently updated documents.
|
||||
|
||||
Example:
|
||||
count = MongoQueryUtils.count_recent_updates(
|
||||
collection=mongo_db['series'],
|
||||
minutes=60,
|
||||
)
|
||||
print(f"{count} series were updated in the last hour")
|
||||
"""
|
||||
if query_filter is None:
|
||||
query_filter = {}
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
cutoff_time = now - timedelta(minutes=minutes)
|
||||
cutoff_epoch = int(cutoff_time.timestamp())
|
||||
|
||||
query = {
|
||||
**query_filter,
|
||||
'$or': [
|
||||
{updated_field: {'$gte': cutoff_epoch}},
|
||||
{updated_field: {'$gte': cutoff_time.isoformat()}},
|
||||
]
|
||||
}
|
||||
|
||||
count = collection.count_documents(query)
|
||||
return count
|
||||
|
||||
|
||||
# Convenience functions
|
||||
def get_recent_updates(
|
||||
collection: Any,
|
||||
minutes: int = 60,
|
||||
updated_field: str = 'updated',
|
||||
query_filter: Optional[Dict] = None,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Find documents updated in the last N minutes."""
|
||||
return MongoQueryUtils.get_recent_updates(
|
||||
collection=collection,
|
||||
minutes=minutes,
|
||||
updated_field=updated_field,
|
||||
query_filter=query_filter,
|
||||
)
|
||||
|
||||
|
||||
def get_recent_ids(
|
||||
collection: Any,
|
||||
minutes: int = 60,
|
||||
updated_field: str = 'updated',
|
||||
id_field: str = 'id',
|
||||
query_filter: Optional[Dict] = None,
|
||||
) -> List[int]:
|
||||
"""Get IDs of documents updated in the last N minutes."""
|
||||
return MongoQueryUtils.get_recent_updates_by_id(
|
||||
collection=collection,
|
||||
minutes=minutes,
|
||||
updated_field=updated_field,
|
||||
id_field=id_field,
|
||||
query_filter=query_filter,
|
||||
)
|
||||
|
||||
|
||||
def count_recent(
|
||||
collection: Any,
|
||||
minutes: int = 60,
|
||||
updated_field: str = 'updated',
|
||||
query_filter: Optional[Dict] = None,
|
||||
) -> int:
|
||||
"""Count documents updated in the last N minutes."""
|
||||
return MongoQueryUtils.count_recent_updates(
|
||||
collection=collection,
|
||||
minutes=minutes,
|
||||
updated_field=updated_field,
|
||||
query_filter=query_filter,
|
||||
)
|
||||
Reference in New Issue
Block a user