feed.database
Documentation for eth_defi.feed.database Python module.
DuckDB persistence for vault post tracking.
Module Attributes
DuckDB storage version for new vault-post databases. |
|
Vault-post tables owned by |
|
Columns owned by each vault-post table, in the migration target's order. |
|
Nullable columns added after the original post-scanner schema. |
|
Per-column storage inspection queries for Zstandard-compressed post content. |
Functions
Resolve the vault post feed DuckDB database path. |
Classes
A single normalised post ready for database insertion. |
|
DuckDB database for tracked sources and collected posts. |
- DEFAULT_VAULT_POST_DATABASE = PosixPath('/home/runner/.tradingstrategy/vaults/vault-post-database.duckdb')
Default DuckDB path for collected vault posts.
- VAULT_POST_DUCKDB_STORAGE_COMPATIBILITY_VERSION = 'latest'
DuckDB storage version for new vault-post databases.
latestenables native Zstandard compression for long post text and raw source API payloads. The resulting file requires its writer’s DuckDB version or a newer version.
- VAULT_POST_DATABASE_TABLES = ('tracked_sources', 'posts', 'feed_sync_state')
Vault-post tables owned by
VaultPostDatabaseand copied during migration.
- VAULT_POST_DATABASE_COLUMNS = {'feed_sync_state': ('key', 'value', 'updated_at'), 'posts': ('source_id', 'external_post_id', 'title', 'post_url', 'published_at', 'fetched_at', 'short_description', 'full_text', 'ai_summary', 'raw_payload'), 'tracked_sources': ('source_id', 'feeder_id', 'name', 'role', 'website', 'source_type', 'source_key', 'canonical_url', 'last_checked_at', 'last_success_at', 'last_error', 'last_post_published_at', 'added_at', 'updated_at')}
Columns owned by each vault-post table, in the migration target’s order.
Existing database files can predate a nullable column added through
ALTER TABLE. Migration copies by these names rather than source column position, preserving the historical schema variants.
- VAULT_POST_OPTIONAL_MIGRATION_COLUMNS = frozenset({'raw_payload', 'website'})
Nullable columns added after the original post-scanner schema.
- VAULT_POST_COMPRESSION_QUERIES = {'full_text': "\n SELECT DISTINCT compression\n FROM pragma_storage_info('posts')\n WHERE column_name = 'full_text'\n AND segment_type = 'VARCHAR'\n AND persistent\n ", 'raw_payload': "\n SELECT DISTINCT compression\n FROM pragma_storage_info('posts')\n WHERE column_name = 'raw_payload'\n AND segment_type = 'VARCHAR'\n AND persistent\n "}
Per-column storage inspection queries for Zstandard-compressed post content.
- resolve_feed_database_path()
Resolve the vault post feed DuckDB database path.
Mirrors the resolution used by the post scanner (
scripts/erc-4626/scan-vault-posts.py) so the JSON export reads the same database the feed collector writes. Keeping the resolver next toDEFAULT_VAULT_POST_DATABASEavoids each caller repeating the same environment lookup and path expansion logic.The
FEED_DB_PATHoverride takes precedence, falling back to theDB_PATHvariable consumed by the post scanner, then the default path.- Returns
Path from
FEED_DB_PATH, thenDB_PATH, then the default vault post database path.- Return type
- class CollectedPost
Bases:
objectA single normalised post ready for database insertion.
- external_post_id: str
Stable external identifier derived from the feed entry or a deterministic fallback.
- published_at: Optional[datetime.datetime]
Original post publication timestamp in naive UTC.
- fetched_at: datetime.datetime
Timestamp when the collector fetched this post in naive UTC.
- short_description: str
Short preview text stored alongside the post.
Capped at 200 characters for compact listings; this is not the complete post body. See
full_textfor the full content.
- full_text: str
Best available full text extracted from the feed entry.
For X/Twitter this is the complete note tweet body for tweets longer than 280 characters, populated via
eth_defi.feed.twitter_api._extract_full_tweet_text(). Seeshort_descriptionfor the truncated preview.
- raw_payload: Optional[str]
JSON-serialised raw payload from the source API (e.g. full tweet object from X API).
- __init__(external_post_id, title, post_url, published_at, fetched_at, short_description, full_text, ai_summary=None, raw_payload=None)
- class VaultPostDatabase
Bases:
objectDuckDB database for tracked sources and collected posts.
- __init__(path)
- Parameters
path (pathlib.Path) –
- close()
Close the database connection.
- Return type
None
- save()
Force a checkpoint.
- Return type
None
- upsert_tracked_source(source)
Insert or update one tracked source and return its source ID.
- Parameters
source (eth_defi.feed.sources.TrackedPostSource) –
- Return type
- upsert_tracked_sources(sources)
Insert or update tracked sources and return source IDs by logical key.
- Parameters
sources (collections.abc.Iterable[eth_defi.feed.sources.TrackedPostSource]) –
- Return type
- mark_source_success(source_id, *, checked_at=None, last_post_published_at=None)
Update sync state for a successful source fetch.
- Parameters
source_id (int) –
checked_at (Optional[datetime.datetime]) –
last_post_published_at (Optional[datetime.datetime]) –
- Return type
None
- mark_source_failure(source_id, error, *, checked_at=None)
Update sync state for a failed or skipped source fetch.
- Parameters
source_id (int) –
error (str) –
checked_at (Optional[datetime.datetime]) –
- Return type
None
- insert_posts(source_id, posts)
Insert posts for a source and return the number of new rows.
- Parameters
source_id (int) –
posts (collections.abc.Iterable[eth_defi.feed.database.CollectedPost]) –
- Return type
- prune_posts(max_post_age_days)
Delete posts older than the configured retention period.
- get_sync_state(key)
Read a value from the feed_sync_state table.
- set_sync_state(key, value)
Write a value to the feed_sync_state table.
- get_known_post_ids(source_id=None)
Return all known external_post_id values, optionally filtered by source.
- get_source_last_post_timestamps(source_ids)
Return the stored
last_post_published_atfor the given source IDs.Used to gate backfill fallbacks: a source whose stored timestamp is not
Nonehas already been seen before and does not need a fallback individual timeline read.- Parameters
source_ids (collections.abc.Iterable[int]) – Iterable of numeric source IDs to look up.
- Returns
Mapping of
source_id → last_post_published_at(Nonewhen the column has never been set for that row).- Return type
dict[int, datetime.datetime | None]
- get_tracked_sources_df()
Return tracked source rows for diagnostics.
- Return type
- get_posts_df()
Return stored posts for diagnostics.
- Return type
- fetch_recent_posts_by_feeder(feeder_ids, max_per_feeder=10)
Fetch the most recent posts for each feeder across all source types.
Joins
tracked_sourcesandpostsonsource_id, ranks posts per feeder byCOALESCE(published_at, fetched_at) DESC, and returns the max_per_feeder newest posts per feeder.- Parameters
feeder_ids (collections.abc.Iterable[str]) – Iterable of feeder-id slugs to look up.
max_per_feeder (int) – Maximum number of posts to return per feeder.
- Returns
Dict mapping
feeder_idto a list of post dicts with keystitle,short_description,full_text,post_url,source_type,published_at(always set via COALESCE fallback tofetched_at). Lists are ordered newest-first.- Return type