warren.storage.documents
Documents: document locations, resolvers, the cached fetcher and source discovery.
resolve_gcs needs the warren[gcs] extra, resolve_s3 needs the
warren[s3] extra, and the HTTP names need the warren[http] extra. The
package imports them only when you use them, so the package imports without
these extras.
Fetch documents
GetDocumentFunc
protocolGetDocumentFuncmethod__call__async (doc_id, document_location, *, job_id=None) -> bytesFetch document bytes by location, with transparent caching.
Workers call this to obtain raw document bytes. The implementation handles cache lookup, resolution dispatch, and cache population.
paramdoc_idstrDocument identifier (used as cache key).
paramdocument_locationDocumentLocationWhere the document lives.
paramjob_idstr | None= NoneJob identifier scoping the cache entry. Workers should pass message.job_id so a re-submitted doc_id with updated content never reads bytes cached by a previous job. None falls back to the legacy shared per-doc key.
Returns
bytesRaw document bytes.
Raises:
DocumentNotFoundError: If the document does not exist.UnknownLocationTypeError: If no resolver for the location type.DocumentResolutionError: For transient resolution failures. The consumer does not retry it: raiseSoftFailureExceptionto retry.
CachedDocumentFetcher
classCachedDocumentFetcher(*, cache, resolvers, name=None, cache_ttl_seconds=86400)Fetches document bytes with cache-aside pattern.
Uses a binary cache (e.g., RedisBinaryCache) for caching and a dispatch table of resolver functions for location-type-specific fetching.
paramcacheCacheInterface[bytes]Binary cache instance for storing resolved bytes.
paramresolversMapping[str, ResolveDocumentFunc]Mapping of location_type to resolver function.
paramnamestr | None= NoneOptional name for logging.
paramcache_ttl_secondsint= 86400Expiry for both shared and job-scoped keys.
method__call__async (doc_id, document_location, *, job_id=None) -> bytesFetch document bytes, using cache when available.
paramdoc_idstrDocument identifier (used as cache key).
paramdocument_locationDocumentLocationWhere the document lives.
paramjob_idstr | None= NoneJob identifier scoping the cache entry. Pass it so a re-submitted doc_id with updated content (a new job) never hits bytes cached by a previous job; None keeps the legacy shared doc:{doc_id} key.
Returns
bytesRaw document bytes.
Raises:
UnknownLocationTypeError: If no resolver for the location type.DocumentNotFoundError: If the document doesn't exist at the location.DocumentResolutionError: For transient resolution failures.CacheOperationError: If the cache read or write fails (not wrapped).HardFailureException: If the cache is scoped and the current handler has no valid scope.
create_cached_document_fetcher
functioncreate_cached_document_fetcher(*, redis_client, resolvers, cache_base_key='documents', default_ttl_seconds=DEFAULT_TTL_SECONDS, scoping_enabled=False, scope_required=True) -> CachedDocumentFetcherCreate a CachedDocumentFetcher with a RedisBinaryCache backend.
paramredis_clientRedisAsync Redis client instance.
paramresolversMapping[str, ResolveDocumentFunc]Mapping of location_type to resolver function.
paramcache_base_keystr= 'documents'Redis key namespace for document cache.
paramdefault_ttl_secondsint= DEFAULT_TTL_SECONDSDefault TTL for cached document bytes. Must be positive, or ValueError.
paramscoping_enabledbool= FalsePrefix each cache key with the scope of the current handler: s:{scope}:{base_key}:{key}.
paramscope_requiredbool= TrueWith scoping on, raise HardFailureException when the handler has no scope. False: use the key without a prefix.
Returns
CachedDocumentFetcherConfigured CachedDocumentFetcher instance.
build_document_cache_key
functionbuild_document_cache_key(doc_id, job_id=None) -> strCompute the cache key a document's bytes live under.
An upstream worker (for example through BinaryResultsStore) can
write bytes to the cache under this key. The downstream
CachedDocumentFetcher reads the same key, so it does not call a
resolver for bytes that an earlier worker already fetched.
When job_id is given the key is job-scoped
(doc:{doc_id}:{job_id}): retries of the same job still hit the
cache, while a re-submission of the same doc_id with new content
(a new job) always fetches fresh bytes instead of serving stale
cached ones. Without job_id the legacy shared key
(doc:{doc_id}) is used — subject to cross-job staleness.
paramdoc_idstrparamjob_idstr | None= NoneDEFAULT_TTL_SECONDS
constantDEFAULT_TTL_SECONDS: int = 86400Locations
DocumentLocation
modelDocumentLocationBase class for document locations.
Subclasses declare a location_type literal that serves as the
Pydantic discriminator for polymorphic deserialization.
attributelocation_typestrDocumentPathLocation
modelDocumentPathLocationDocument on the local filesystem.
attributelocation_typeLiteral['path']= 'path'attributerelative_pathstrPath relative to a configurable base directory.
DocumentURLLocation
modelDocumentURLLocationDocument at an HTTP(S) URL.
attributelocation_typeLiteral['url']= 'url'attributeurlstrFull URL to the document.
DocumentCloudLocation
modelDocumentCloudLocationDocument in cloud object storage (S3, GCS).
attributelocation_typeLiteral['cloud']= 'cloud'attributeproviderLiteral['s3', 'gcs']Cloud provider identifier.
attributebucketstrBucket or container name.
attributekeystrObject key (path within the bucket).
attributeregionstr | None= NoneOptional region for the bucket.
AnyDocumentLocation
type aliasAnyDocumentLocation = Annotated[DocumentPathLocation | DocumentURLLocation | DocumentCloudLocation, Field(discriminator='location_type')]Discriminated union type for deserializing any DocumentLocation from a dict.
Resolvers
ResolveDocumentFunc
type aliasResolveDocumentFunc = Callable[[DocumentLocation], Awaitable[bytes]]Async function that fetches bytes for a specific location type.
One resolver per location_type. Bound with config (e.g., base_dir)
via functools.partial at wiring time.
Resolvers should raise:
DocumentNotFoundErrorfor permanent issues (file missing, HTTP 404).- Let transient exceptions propagate (connection errors, timeouts) —
the fetcher wraps them in
DocumentResolutionError.
resolve_path
functionresolve_pathasync (location, *, base_dir) -> bytesRead document bytes from a file confined to base_dir.
paramlocationDocumentLocationMust be a DocumentPathLocation.
parambase_dirPathDirectory every read is confined to. Absolute paths, .. traversal and symlinks resolving outside it are refused.
Returns
bytesRaw document bytes.
Raises:
DocumentNotFoundError: If the path is missing or escapesbase_dir.
resolve_s3
functionresolve_s3async (location, *, client) -> bytesRead document bytes from Amazon S3.
paramlocationDocumentLocationMust be a DocumentCloudLocation with provider="s3".
paramclientAnyboto3 S3 client instance, bound via functools.partial at wiring time.
Returns
bytesRaw document bytes.
Raises:
DocumentNotFoundError: If the object or bucket does not exist.
resolve_gcs
functionresolve_gcsasync (location, *, client) -> bytesRead document bytes from Google Cloud Storage.
paramlocationDocumentLocationMust be a DocumentCloudLocation with provider="gcs".
paramclientstorage.ClientGCS client instance, bound via functools.partial at wiring time.
Returns
bytesRaw document bytes.
Raises:
DocumentNotFoundError: If the object does not exist.
resolve_http
functionresolve_httpasync (location, *, client) -> bytesRead document bytes from an HTTP(S) URL.
paramlocationDocumentLocationMust be a DocumentURLLocation.
paramclienthttpx.AsyncClienthttpx.AsyncClient instance, bound via functools.partial at wiring time.
Returns
bytesRaw document bytes.
Raises:
DocumentNotFoundError: If the server responds 404 (hard failure).DocumentThrottledError: On 429, or 503 carrying Retry-After (soft failure).
build_client
functionbuild_client(*, follow_redirects=None, timeout_s=None, max_redirects=None, transport=None) -> httpx.AsyncClientConstruct the AsyncClient the URL resolver is bound to.
Explicit arguments win; anything left as None falls back to the
environment, then to the module defaults.
paramfollow_redirectsbool | None= NoneOverride for HTTP_FOLLOW_REDIRECTS.
paramtimeout_sfloat | None= NoneOverride for HTTP_TIMEOUT_S.
parammax_redirectsint | None= NoneOverride for HTTP_MAX_REDIRECTS.
paramtransporthttpx.AsyncBaseTransport | None= NoneSubstitute transport. Exists so the redirect policy can be exercised against httpx.MockTransport without a network or a reach into httpx's private attributes.
Returns
httpx.AsyncClientA configured httpx.AsyncClient.
Raises:
ValueError: If an environment override cannot be parsed.
parse_retry_after
functionparse_retry_after(value, *, now=None) -> float | NoneSeconds asked for by a Retry-After header (delay-seconds or HTTP-date).
paramvaluestr | NoneRaw header value, or None when the header is absent.
paramnowdatetime | None= NoneReference instant for an HTTP-date; defaults to the clock.
Returns
float | NoneNon-negative seconds, or None when absent or unparseable.
DEFAULT_FOLLOW_REDIRECTS
constantDEFAULT_FOLLOW_REDIRECTS: bool = TrueDEFAULT_MAX_REDIRECTS
constantDEFAULT_MAX_REDIRECTS: int = 20DEFAULT_TIMEOUT_S
constantDEFAULT_TIMEOUT_S: float = 60.0Errors
DocumentResolutionError
errorDocumentResolutionError(message, *, doc_id=None)Base exception for document resolution failures.
Not retried by itself: the consumer treats any exception other than
SoftFailureException as a hard failure. To retry, catch this error
in the handler and raise SoftFailureException.
parammessagestrError description.
paramdoc_idstr | None= NoneDocument identifier, if known.
DocumentNotFoundError
errorDocumentNotFoundError(message, *, doc_id=None)Document does not exist at the specified location.
Do not retry it: the document will not appear on retry.
Raised by resolvers and the fetcher. Document stores raise a different
class, warren.storage.document_store.DocumentNotFoundError. A handler
that catches one class does not catch the other.
parammessagestrparamdoc_idstr | None= NoneUnknownLocationTypeError
errorUnknownLocationTypeError(message, *, doc_id=None)No resolver registered for the given location type.
Do not retry it: a missing resolver is a configuration error, not a transient issue.
parammessagestrparamdoc_idstr | None= NoneDocumentThrottledError
errorDocumentThrottledError(message, *, retry_after, status_code, doc_id=None)The source asked us to slow down: HTTP 429, or 503 with Retry-After.
Not retried by itself, as for DocumentResolutionError. To retry
after the delay that the source asked for, catch this error in the
handler and raise SoftFailureException with retry_after.
parammessagestrError description.
paramretry_afterfloat | NoneServer-requested delay in seconds, or None when the response carried no usable Retry-After.
paramstatus_codeintThe HTTP status that carried the request.
paramdoc_idstr | None= NoneDocument identifier, if known.
Source discovery
discover_local_pdfs
functiondiscover_local_pdfs(pdf_dir, max_docs=0) -> Sequence[Path]Find PDF files in a local directory.
parampdf_dirPathdirectory to search for PDFs.
parammax_docsint= 0maximum number of PDFs to return (0 = all).
Returns
Sequence[Path]sorted list of PDF file paths.
Raises:
FileNotFoundError: if the directory doesn't exist.ValueError: if no PDFs are found.
discover_gcs_pdfs
functiondiscover_gcs_pdfs(bucket_name, prefix='pdfs/', max_docs=0) -> Sequence[str]List PDF objects in a GCS bucket.
parambucket_namestrGCS bucket name.
paramprefixstr= 'pdfs/'object key prefix (e.g., "pdfs/").
parammax_docsint= 0maximum number of objects to return (0 = all).
Returns
Sequence[str]sorted list of object keys.
Raises:
OptionalDependencyError: ifgoogle-cloud-storageis not installed.ValueError: if no PDFs are found.