Module

content_layer.source

ContentSource - Abstract base class for content sources.

Defines the protocol that all content sources must implement.

View source
1 Class

Classes

ContentSource abstract
Abstract base class for content sources. Implementations fetch content from various origins (local…
10

Abstract base class for content sources.

Implementations fetch content from various origins (local files, remote APIs, databases) and return unified ContentEntry objects.

Subclasses must implement:

  • source_type property
  • fetch_all() async generator
  • fetch_one() async method
Inherits from ABC

Methods 8

source_type property
Return source type identifier.
str 
def source_type(self) -> str

Return source type identifier.

Returns

str —

String identifier like 'local', 'github', 'notion', 'rest'

fetch_all async
Fetch all content entries from this source.
0 AsyncIterator[Conte… 
async def fetch_all(self) -> AsyncIterator[ContentEntry]

Fetch all content entries from this source.

Returns

AsyncIterator[ContentEntry]

fetch_one async
Fetch a single content entry by ID.
1 ContentEntry | None 
async def fetch_one(self, id: str) -> ContentEntry | None

Fetch a single content entry by ID.

Parameters 1
id str

Source-specific identifier (e.g., file path, doc ID)
Returns

ContentEntry | None —

ContentEntry if found, None otherwise

get_cache_key
Generate cache key for this source configuration. Used to invalidate cache whe…
0 str 
def get_cache_key(self) -> str

Generate cache key for this source configuration.

Used to invalidate cache when config changes. Override for custom cache key logic.

Returns

str —

16-character hex string based on config hash

get_last_modified async
Get last modification time for the entire source. Used for cache invalidation.…
0 datetime | None 
async def get_last_modified(self) -> datetime | None

Get last modification time for the entire source.

Used for cache invalidation. Returns None if unknown or not supported.

Returns

datetime | None —

Last modification datetime or None

is_changed async
Check if source content has changed since last fetch.
1 bool 
async def is_changed(self, cached_checksum: str | None) -> bool

Check if source content has changed since last fetch.

Parameters 1
cached_checksum str | None

Previously cached checksum
Returns

bool —

True if content may have changed, False if definitely unchanged

fetch_all_sync
Synchronous wrapper for fetch_all(). Runs the async generator in a new event loop.
0 Iterator[ContentEntry] 
def fetch_all_sync(self) -> Iterator[ContentEntry]

Synchronous wrapper for fetch_all().

Runs the async generator in a new event loop.

Returns

Iterator[ContentEntry]

fetch_one_sync
Synchronous wrapper for fetch_one().
1 ContentEntry | None 
def fetch_one_sync(self, id: str) -> ContentEntry | None

Synchronous wrapper for fetch_one().

Parameters 1
id str

Source-specific identifier
Returns

ContentEntry | None —

ContentEntry if found, None otherwise

Internal Methods 2
__init__
Initialize content source.
2 None 
def __init__(self, name: str, config: dict[str, Any]) -> None

Initialize content source.

Parameters 2
name str

Unique name for this source instance (e.g., "api-docs")
config dict[str, Any]

Source-specific configuration dictionary
__repr__
0 str 
def __repr__(self) -> str
Returns

str