"""Provides GRR protocols based on fsspec library."""
# pylint: disable=too-many-lines
from __future__ import annotations
import abc
import asyncio
import base64
import copy
import datetime
import gzip
import hashlib
import json
import operator
import os
import pathlib
import tempfile
import time
import uuid
from collections.abc import Callable, Generator, Iterable
from contextlib import AbstractContextManager, contextmanager, suppress
from dataclasses import asdict, dataclass
from threading import Event, Lock, get_ident
from typing import (
IO,
Any,
NamedTuple,
cast,
)
from urllib.parse import urlparse
from weakref import finalize, ref
import apsw
import fsspec
import fsspec.exceptions
import pathspec
import pyBigWig
import pysam
import yaml
from filelock import FileLock
from gain import logging
from gain.genomic_resources.dvc import (
DVC_SUFFIX,
UnsupportedDvcDirectoryOutputError,
parse_dvc_pointer_out,
)
from gain.genomic_resources.repository import (
GR_CONF_FILE_NAME,
GR_CONTENTS_FILE_NAME,
GR_INDEX_FILE_NAME,
GR_LEGACY_CONTENTS_FILE_NAME,
GR_MANIFEST_FILE_NAME,
GR_SQLITE_META_FILE_NAME,
GenomicResource,
Manifest,
ManifestEntry,
Mode,
ReadOnlyRepositoryProtocol,
ReadWriteRepositoryProtocol,
ResourceFileState,
ResourceScan,
SearchIndexUnavailableError,
escape_unsafe_characters,
is_generated_info_page,
is_gr_id_token,
malformed_resource_id_reason,
parse_gr_id_version_token,
resolve_tabix_index_filename_for_read,
uncontained_resource_id_reason,
validate_resource_file_name,
validate_resource_id,
)
from gain.templates import get_template
from gain.templates.markdown_support import render_markdown as markdown
from gain.templates.static_assets import repository_static_files
from gain.utils.fs_utils import S3_PRESIGN_EXPIRATION_SECONDS
from gain.utils.helpers import convert_size
from gain.utils.resource_id_order import resource_id_collation_key
from gain.utils.url_redaction import (
strip_url_credentials,
strip_url_userinfo,
)
# Silence the spurious "[W::hts_idx_load3] The index file is older than the
# data file" warning that htslib emits when a tabix/VCF index has an older
# mtime than its data file. In our GRR workflow this is benign: both the
# caching protocol and DVC download index and data files in parallel, and
# the smaller index typically lands first. Level 1 (errors only) keeps real
# htslib errors visible while suppressing notices (3) and warnings (2).
pysam.set_verbosity(1)
logger = logging.getLogger(__name__)
def _declare_index_contigs(vcf_file: pysam.VariantFile) -> None:
"""Add the index's contigs to a header that does not declare them.
htslib does this itself -- but only for an index it found by ITSELF,
probing next to the data file: ``vcf_hdr_read`` loads an adjacent
``.tbi``/``.csi`` and folds the names it lists into the header. An
index opened by an explicit path (every remote open, and every open of
a table whose ``index_filename`` names a non-adjacent file) misses that
step, and a VCF with no ``##contig`` lines then has an EMPTY header
contig list -- so ``fetch("chr1")`` raises ``ValueError: invalid contig
chr1`` although the index knows the contig perfectly well.
Declaring them here is what makes an explicitly-indexed open behave
exactly like an auto-probed one, which is the whole point of honouring a
configured index (gain#596). A contig the header already declares is
left alone.
"""
index = vcf_file.index
if index is None:
return
contigs = vcf_file.header.contigs
for indexed in index:
contig = str(indexed)
if contig not in contigs:
contigs.add(contig)
[docs]
class FileCacheVerdict(NamedTuple):
"""The lock-free classification of a single resource file.
``needs_download`` is True when the local copy is missing or has drifted
from the remote manifest and must be (re)downloaded; ``size`` is the
manifest-recorded byte size of that pending download (0 when nothing
needs downloading). See gain#78.
"""
needs_download: bool
size: int
# Per-file download retry policy for copy_resource_file. A single stalled or
# dropped read over a slow HTTP GRR link used to abort the whole cache run
# (gain#43); instead we retry the file from scratch with exponential backoff.
_COPY_MAX_ATTEMPTS = 4
_COPY_BACKOFF_BASE = 5 # seconds; delays are 5s, 15s, 45s
#: Directory inside a resource holding the protocol's own bookkeeping --
#: per-file ``.state`` documents, lockfiles, and partial downloads. Not part
#: of the resource: everything that enumerates resource files skips it.
GRR_INTERNAL_DIR = ".grr"
[docs]
class RetryableCopyError(OSError):
"""A copy failure the download loop retries from scratch.
Retryability is a property of the class, not of a list kept beside it:
``copy_resource_file`` catches this base, so a new transient failure
shape becomes retryable by subclassing it and nothing else (gain#934).
What the retry buys is a fresh remote handle and a fresh temp file, so
only faults a second full attempt could plausibly clear belong here --
a stalled link, a short read, corrupt bytes. A fault that would repeat
identically is a plain ``OSError`` and surfaces on the first attempt.
The classification is the *download* path's, and it reaches
everything that path does -- the move included, so a corrupting
publish inside a download is retried like any other transient fault.
One shape reaches here from outside a download: redacting a
credential-bearing failure whose own type cannot be reconstructed from a
message rebuilds it as this class, so that redaction cannot silently
reclassify a transient failure as permanent (gain#1078, ADR 0023). Such
an error can therefore surface from a plain read -- ``get_file_content``,
a tabix header -- where nothing will retry it. It is still true of it
that a second attempt could plausibly have cleared it; only the
opportunity is absent.
"""
[docs]
class ChecksumMismatchError(RetryableCopyError):
"""A completed download whose md5 disagrees with the manifest.
Almost always a truncated or corrupted transfer.
"""
[docs]
class TruncatedDownloadError(RetryableCopyError):
"""A download that ended short of the manifest's recorded byte size.
A silent short read in the fsspec range-reassembly layer (gain#292, H1)
makes ``infile.read()`` return EOF before the whole file has been
streamed; the copy loop stops on that empty read and writes a truncated
file. Caught explicitly by byte count -- before the md5 check -- so the
failure is reported as the truncation it is, with both the received and
the expected size, rather than as an opaque checksum mismatch.
"""
[docs]
class CorruptedPublishError(RetryableCopyError):
"""A published object that differs in size from the verified download.
The download's byte-count and md5 checks both run before the move, so
they describe the temp file, not the object the move landed. Since the
digest is carried across rather than recomputed from the store
(gain#865), a move that published something else -- a partial
object-store copy, a rename that lost the tail, one that left no
object at all -- would be recorded under the digest of bytes that are
no longer there: size and timestamp come from the published object and
agree with it, the md5 comes from the download and agrees with the
manifest, so every later cache verdict passes and the file is served
corrupt indefinitely (gain#880).
Only size-changing corruption is caught. Catching the rest needs a
second full read of every downloaded file -- for a remote store, a
second transfer -- which is exactly the cost gain#865 removed.
Raised on a second, digest-free path since gain#933: the repository's
own artifacts publish through :meth:`_publish_file`, which compares
the published object against the bytes it staged rather than against
a manifest. The reasoning above is about the download, but the guard
is the same one -- it lives on the move, so everything that moves
gets it. The retryability inherited from
:class:`RetryableCopyError` does not reach that second path, though:
its callers publish once and propagate, so a corrupting publish of a
repository artifact fails rather than retrying.
"""
# aiohttp.ClientError is folded into the retryable set when aiohttp is
# importable (it always is when an HTTP GRR is used).
try:
import aiohttp as _aiohttp
_aiohttp_errors: tuple[type[BaseException], ...] = (_aiohttp.ClientError,)
except ImportError:
_aiohttp_errors = ()
# Transient errors that warrant retrying a file download from scratch. A
# stalled aiohttp read surfaces as fsspec's FSTimeoutError, and ConnectionError
# covers resets/refused connects; the tuple exists for those -- failure types
# gain does not own and so cannot give a base class. Every gain-owned shape
# enters through RetryableCopyError instead, which is why this list does not
# grow when one is added.
_RETRYABLE_COPY_ERRORS: tuple[type[BaseException], ...] = (
fsspec.exceptions.FSTimeoutError,
asyncio.TimeoutError,
ConnectionError,
RetryableCopyError,
*_aiohttp_errors,
)
def _strip_netloc_userinfo(netloc: str) -> str:
"""Return a network location with any ``user:pass@`` userinfo removed.
The authority (``host[:port]``) is kept verbatim — case, port and IPv6
brackets are preserved — because only the userinfo carries the secret. A
netloc without ``@`` is returned unchanged. Splitting on the LAST ``@`` is
correct for well-formed urls: the host part never contains an unencoded
``@`` (a literal ``@`` inside userinfo must be percent-encoded as ``%40``).
"""
at_index = netloc.rfind("@")
if at_index == -1:
return netloc
return netloc[at_index + 1:]
def _display_url(url: str) -> str:
"""Return the credential-free ``scheme://netloc/path`` form of a url.
One definition of a protocol's display identity, used both to derive
``self.url`` and to decide whether a rebuild asking for the default
``public_url`` is asking for the incumbent's (#514).
"""
parsed = urlparse(url)
scheme = parsed.scheme or "file"
return f"{scheme}://{_strip_netloc_userinfo(parsed.netloc)}{parsed.path}"
def _fetch_url_form(url: str) -> str:
"""Return the credential-BEARING ``scheme://netloc/path`` form of a url.
``_display_url``'s counterpart: identical to it for a userinfo-free url,
and the form every remote read must derive from when the url does carry
``user:pass@`` (see ``FsspecReadOnlyProtocol._fetch_url``). It is also the
form the protocol memo is keyed on, so a caller that passes a bare
``/abs/path`` and one that passes its ``file:///abs/path`` spelling name
one protocol -- and a pickle, which carries the ``file://`` form, lands
back on the instance it came from (#514).
"""
parsed = urlparse(url)
scheme = parsed.scheme or "file"
return f"{scheme}://{parsed.netloc}{parsed.path}"
def _rebuild_error_without_url_credentials(
exc: BaseException, redacted: str) -> BaseException:
"""Rebuild ``exc`` carrying the ``redacted`` message instead of its own.
Return a fresh exception of ``exc``'s type so it can propagate/log
without leaking the secret its own message carries; fall back for a type
that cannot be reconstructed from a single message string.
That fallback loses the exception's identity, and with it any decision a
caller makes by type. The one such decision in the tree is the download
loop's ``_RETRYABLE_COPY_ERRORS`` classification, so it is preserved
explicitly: a transient failure stays transient across redaction.
Before gain#1078 that was a positional guarantee -- ``copy_resource_file``
redacted on the way out, strictly after its own ``except`` had classified
(gain#620) -- and redacting "any earlier" was called out in that loop as
something that would quietly cut the retry budget to a single attempt for
exactly the authed downloads it protects. Redacting the handle itself IS
earlier: it runs under the loop, on the read. Making retryability survive
the rebuild is what lets it, and turns a rule about *where* redaction may
sit into a property of the rebuild.
"""
try:
return type(exc)(redacted)
except Exception: # ruff: ignore[blind-except] # pylint: disable=broad-exception-caught
if isinstance(exc, _RETRYABLE_COPY_ERRORS):
# ``RetryableCopyError`` is how this tree says "transient"
# (gain#934): a new transient shape becomes retryable by
# subclassing it and nothing else. A redacted rebuild of a
# transient failure is such a shape.
return RetryableCopyError(redacted)
return OSError(redacted)
def _error_without_url_credentials(exc: BaseException) -> BaseException:
"""Return a credential-free rebuild of ``exc``, or ``exc`` if it has none.
"Credential" is whatever :func:`strip_url_credentials` recognises --
``user:pass@`` userinfo and a presigned url's query string alike -- so a
message carrying both loses both, in one rebuild.
Returning ``exc`` itself is what every failure carrying no credential
wants -- the common unauthenticated case -- because a rebuild always
costs the traceback, and costs the exception its type too whenever it
cannot be reconstructed from a message alone. Callers can therefore
raise the result unconditionally, and identity is what tells them
whether anything was redacted.
"""
message = str(exc)
redacted = strip_url_credentials(message)
if redacted == message:
return exc
return _rebuild_error_without_url_credentials(exc, redacted)
def _run_redacting_url_credentials[T](fn: Callable[[], T]) -> T:
"""Run ``fn``, re-raising any failure with its url credentials stripped.
On a fetch failure fsspec/aiohttp embed the credential-bearing fetch url
verbatim in the raised message (e.g. ``FileNotFoundError(url)``). Rebuild
the error with the url's credentials stripped and raise it OUTSIDE the
``except`` block so no credential-bearing ``__context__``/``__cause__``
survives a chain walk. A failure whose message carries no credential (the
common non-authed case) is propagated unchanged.
Both shapes are stripped here, not just userinfo: an s3 GRR is handed a
presigned url whose signature is a query parameter, and it reaches this
guard by exactly the same routes (gain#1339).
"""
reraise: BaseException | None = None
try:
return fn()
except Exception as exc:
reraise = _error_without_url_credentials(exc)
if reraise is exc:
# Nothing was redacted. Propagate in place, so the original
# traceback and any chain it already carries survive.
raise
raise reraise
def _url_carries_userinfo(url: str) -> bool:
"""Whether ``url`` embeds ``user:pass@`` userinfo.
Defined as "the narrow redactor would change it", so that this predicate
and ``strip_url_userinfo`` cannot come to disagree about what counts as
userinfo. It is the userinfo half of ``_url_carries_credential``, which is
the predicate a caller asking "is there a secret in here" wants.
"""
return strip_url_userinfo(url) != url
#: Serialises ``_htslib_silenced`` below, because htslib's verbosity level is
#: process-global and the save/lower/restore sequence is not atomic. Two
#: threads that overlap in it strand the level: the second saves the FIRST
#: one's 0 as its "previous" level and restores that on the way out, leaving
#: htslib silent -- every ``[E::...]`` line from every later tabix, VCF and
#: fasta open discarded -- for the rest of the process's life (gain#1360).
#:
#: This is not a theoretical pool. ``web_api``'s pipeline cache opens
#: pipelines on a ``ThreadedTaskExecutor`` (8 loaders by default), and
#: opening a pipeline brackets every credentialed tabix, VCF and fasta open
#: in it.
#:
#: Not re-entrant: the only bracket is ``_open_htslib_file``'s, around a
#: bare pysam constructor that enters no bracket of its own (ADR 0023's
#: gain#1406 amendment records the second bracket this used to nest with).
#:
#: Separate from ``_STDERR_SUPPRESSION_LOCK`` on purpose: fd 2 and the
#: verbosity level are independent globals, and sharing one lock would
#: serialise credentialed bigwig opens against credentialed htslib opens for
#: nothing.
#:
#: The cost is that credentialed tabix/VCF/fasta opens serialise, each
#: holding the lock across a network open. Accepted for the same reason as
#: the fd 2 lock: it is the open, not the read, so it is not the score-scan
#: hot path, and it is paid only by a GRR whose url carries a credential.
_HTSLIB_VERBOSITY_LOCK = Lock()
@contextmanager
def _htslib_silenced() -> Generator[None]:
"""Run the body at htslib verbosity 0, restoring the previous level.
The previous level is restored rather than assumed: this module sets 1
at import, but the application may have set something else since.
Every ``pysam.set_verbosity(0)`` bracket in the open path goes through
here, so that there is exactly one place that takes the lock (see
``_HTSLIB_VERBOSITY_LOCK`` for why it serialises).
"""
with _HTSLIB_VERBOSITY_LOCK:
saved_verbosity = pysam.set_verbosity(0)
try:
yield
finally:
pysam.set_verbosity(saved_verbosity)
def _open_htslib_file[T](url: str, open_: Callable[[], T]) -> T:
"""Run a pysam open that is handed the credential-bearing ``url``.
``open_tabix_file``, ``open_vcf_file`` and ``open_fasta_file`` hand pysam
a url STRING rather than a handle, so ADR 0023's ``_RedactingFile`` never
sees them -- the library owns the transport, and GAIn composes none of the
messages, so redacting an f-string has nothing to act on either. Both
channels the credential escapes through are closed here (gain#1314).
The credential may be ``user:pass@`` userinfo or, for an s3 GRR whose url
is presigned, a query-string signature. ``_url_carries_credential`` --
gain#1333's predicate, shared rather than copied -- is what makes those
one case rather than two, and gain#1339 is what made it safe to gate on
here: the bracket alone would have silenced htslib for an s3 GRR while
the rebuilt exception still carried the signature.
The error pysam raises embeds the url verbatim, and it is an ``OSError``
-- in ``RESOURCE_ERRORS``, so ``report_resource_failure`` writes its text
to the log at ERROR, where it persists and is shipped.
htslib *additionally* writes the url to fd 2 itself, which no redaction of
the raised exception can reach, so a credential-bearing open is bracketed
at verbosity 0. That silences htslib's own diagnostics for the duration of
the open, which is a genuine loss of detail -- and it is spent only where
it buys something. A GRR that hands over no credential keeps them,
including an ANONYMOUS s3 one, whose ``sign`` returns the url with no
query string at all. The scoping mirrors the type demotion of ADR 0023,
likewise paid by exactly the configuration it protects.
The bracket is ``_htslib_silenced`` -- see it for why it restores rather
than assumes, and ``_HTSLIB_VERBOSITY_LOCK`` for why it serialises and
which pool drives these opens (gain#1360). The equivalent hazard on fd 2
is why ``_open_libbigwig_file`` takes ``_STDERR_SUPPRESSION_LOCK``.
Named for htslib rather than for libraries in general because the
silencing half is ``pysam``-specific: libBigWig has no verbosity control
to bracket, so ``open_bigwig_file`` goes through ``_open_libbigwig_file``
instead.
What this does NOT cover, per ADR 0023's gain#1339 amendment: the
returned handle's later reads. The s3 presigned shape, which the gain#1314
amendment listed here as an open gap, is covered -- the predicate above
recognises it and the redactor strips it. Read the gain#1339 amendment
rather than the gain#1314 one for the current coverage claim.
"""
if not _url_carries_credential(url):
return _run_redacting_url_credentials(open_)
with _htslib_silenced():
return _run_redacting_url_credentials(open_)
def _url_carries_credential(url: str) -> bool:
"""Whether ``url`` carries credential material of any shape.
Two doors, and a bigwig open has to close both. ``user:pass@`` userinfo
is the one an authed http GRR comes through; an s3 GRR comes through the
other, because ``_get_file_url`` presigns for that scheme and the bearer
token then sits in the QUERY STRING with no userinfo at all.
Answers True for any url carrying a ``?``, so an http GRR whose base url
legitimately had a query string -- or a ``file`` GRR under a directory
with a ``?`` in its name -- is treated as credentialed and loses its
libBigWig diagnostics. Deliberate: it costs detail, never correctness.
ADR 0023's gain#1333 amendment records why the test is the query string
whole rather than a parameter-name list or the scheme, both of which were
measured and rejected.
"""
return _url_carries_userinfo(url) or bool(urlparse(url).query)
#: The descriptor libBigWig writes its diagnostics to. Spelled as the number
#: rather than ``sys.stderr.fileno()`` because it is the C-level descriptor
#: that must be replaced, and ``sys.stderr`` may be a Python object bound to
#: something else entirely -- under pytest's ``capfd``, or wherever the
#: application has rebound it.
_STDERR_FILENO = 2
#: Serialises the suppression below, because fd 2 is process-global and the
#: save/redirect/restore sequence is not atomic. Two threads that overlap in
#: it strand the descriptor: the second saves the FIRST one's null device as
#: its "previous" fd 2 and restores that on the way out, leaving the process
#: with no stderr at all for the rest of its life.
#:
#: This is not a theoretical pool. ``web_api``'s pipeline cache loads
#: pipelines on a ``ThreadedTaskExecutor`` (8 loaders by default), and
#: opening a pipeline opens the bigwig tables in it.
#:
#: The cost is that concurrent credentialed bigwig opens serialise. Accepted:
#: it is the open, not the read, so it is not the score-scan hot path, and it
#: is paid only by a GRR whose url carries a credential.
_STDERR_SUPPRESSION_LOCK = Lock()
def _open_libbigwig_file[T](url: str, open_: Callable[[], T]) -> T:
"""Run a pyBigWig open that is handed the credential-bearing ``url``.
Runs ``open_`` with fd 2 pointed at the null device, so libBigWig's
diagnostics -- which name the url verbatim: ``[urlOpen] Couldn't open
<url>`` from the no-curl PyPI wheel, ``[bwOpen] bwg->cl is NULL (<url>)``
from a curl-enabled build -- go nowhere. Only a url that carries a
credential is suppressed; anything else keeps its diagnostics. A url
with no credential is passed to ``_run_redacting_url_credentials`` and
nothing else, so the common path costs no syscalls.
It has to be the file descriptor, not ``sys.stderr`` or
``contextlib.redirect_stderr``: libBigWig writes with its own ``fprintf``
from C, so it holds fd 2 directly and never consults the Python object
those rebind.
Serialised on ``_STDERR_SUPPRESSION_LOCK``, without which two concurrent
opens strand fd 2 at the null device for the life of the process. What
the lock does not buy is isolation -- while it is held, another thread's
stderr goes to the null device too -- which is why the suppression stays
as narrow as one library call.
ADR 0023's gain#1333 amendment records why the gate is wider than
``_open_htslib_file``'s, why this is a separate helper, and what the
serialisation costs.
"""
if not _url_carries_credential(url):
return _run_redacting_url_credentials(open_)
with _STDERR_SUPPRESSION_LOCK:
try:
saved_stderr = os.dup(_STDERR_FILENO)
except OSError:
# fd 2 is closed. There is no diagnostic to suppress and nothing
# to put back, so the open runs unsuppressed rather than failing
# -- a read that worked before this guard existed must not start
# raising because of it.
return _run_redacting_url_credentials(open_)
try:
devnull = os.open(os.devnull, os.O_WRONLY)
try:
os.dup2(devnull, _STDERR_FILENO)
finally:
os.close(devnull)
return _run_redacting_url_credentials(open_)
finally:
os.dup2(saved_stderr, _STDERR_FILENO)
os.close(saved_stderr)
#: Redacted I/O operations that are NOT on every handle, and so must be
#: mirrored from the wrapped object rather than declared. ``readall`` and
#: ``read1`` are absent from fsspec's ``AbstractBufferedFile`` (and so from
#: ``S3File``); ``LocalFileOpener`` additionally has no ``readinto`` or
#: ``readinto1``.
#:
#: Declaring one of these as a method would make ``hasattr`` answer True for
#: a handle that does not have it, and consumers feature-detect: pandas
#: probes for ``read1`` when choosing a read path, takes it, and the
#: forwarding call then dies with ``AttributeError`` on the inner handle. A
#: wrapper must not change what the thing it wraps appears able to do.
_OPTIONAL_REDACTED_OPS = frozenset({
"readall", "readinto", "readinto1", "read1",
})
def _redacting_call(bound: Callable[..., Any]) -> Callable[..., Any]:
"""Wrap one bound method of an inner handle in the redaction guard."""
def call(*args: Any, **kwargs: Any) -> Any:
return _run_redacting_url_credentials(lambda: bound(*args, **kwargs))
return call
class _RedactingFile:
"""A file handle whose failures carry no url credentials.
``_open_fsspec_file`` redacts the open. Everything the caller then does
with the handle -- ``read``, a bounded ``read(n)``, ``readline``,
iteration, ``seek`` -- used to run unwrapped, so on an authed GRR a
failure mid-stream surfaced the credential-bearing fetch url verbatim
(gain#1078). Wrapping the handle is what reaches those: the alternative,
routing each caller through ``_read_fetch_file``, reads the whole file
and so cannot serve the callers that hold the handle for random access,
iterate it lazily, or read it in chunks on purpose.
Transparent by construction: every attribute this class does not name is
delegated, because the handle escapes to ``gzip.open``, ``pandas``,
``json.load`` and the gene-set and gene-model readers, which reach for
``mode``, ``name``, ``closed`` and friends.
Transparency includes what the handle appears *unable* to do. The
operations in ``_OPTIONAL_REDACTED_OPS`` are redacted through
``__getattr__`` rather than declared here, so a handle without them still
answers ``hasattr`` False -- see that constant for what goes wrong
otherwise.
See ADR 0023, which records why redaction belongs on the handle rather
than at each of the 33 call sites that read one.
"""
def __init__(self, inner: IO) -> None:
self._inner = inner
def read(self, *args: Any, **kwargs: Any) -> Any:
return _run_redacting_url_credentials(
lambda: self._inner.read(*args, **kwargs))
def readline(self, *args: Any, **kwargs: Any) -> Any:
return _run_redacting_url_credentials(
lambda: self._inner.readline(*args, **kwargs))
def readlines(self, *args: Any, **kwargs: Any) -> Any:
return _run_redacting_url_credentials(
lambda: self._inner.readlines(*args, **kwargs))
def seek(self, *args: Any, **kwargs: Any) -> Any:
# Seeking a remote file is I/O: fsspec's cached readers fetch the
# block the new offset lands in.
return _run_redacting_url_credentials(
lambda: self._inner.seek(*args, **kwargs))
def tell(self, *args: Any, **kwargs: Any) -> Any:
return _run_redacting_url_credentials(
lambda: self._inner.tell(*args, **kwargs))
def write(self, *args: Any, **kwargs: Any) -> Any:
return _run_redacting_url_credentials(
lambda: self._inner.write(*args, **kwargs))
def flush(self, *args: Any, **kwargs: Any) -> Any:
return _run_redacting_url_credentials(
lambda: self._inner.flush(*args, **kwargs))
def writelines(self, *args: Any, **kwargs: Any) -> Any:
return _run_redacting_url_credentials(
lambda: self._inner.writelines(*args, **kwargs))
def truncate(self, *args: Any, **kwargs: Any) -> Any:
return _run_redacting_url_credentials(
lambda: self._inner.truncate(*args, **kwargs))
def close(self, *args: Any, **kwargs: Any) -> Any:
# A write handle does its store round trip on the way out, so a close
# is as able to surface the fetch url as any read. Every write site in
# this tree reaches that through ``__exit__`` rather than here, which
# is why ``__exit__`` is wrapped too and not merely delegated.
return _run_redacting_url_credentials(
lambda: self._inner.close(*args, **kwargs))
def __iter__(self) -> _RedactingFile:
# Special methods are looked up on the TYPE, never through
# ``__getattr__``, so without these two ``for line in handle`` --
# how the tabular and gene-set readers consume a resource file --
# raises ``TypeError: not iterable`` instead of delegating.
return self
def __next__(self) -> Any:
return _run_redacting_url_credentials(lambda: next(self._inner))
def __enter__(self) -> _RedactingFile:
# The inner handle's own ``__enter__`` answers itself -- that is what
# every file object does -- and the ``with`` body must be handed THIS
# object rather than that one, or every read in the body would run
# unredacted and the wrapper would buy nothing.
_run_redacting_url_credentials(self._inner.__enter__)
return self
def __exit__(self, *exc_info: Any) -> Any:
# Wrapped, not delegated: this is where a ``with`` block releases the
# handle, and it is the path every write site in this tree actually
# takes -- ``with ... as outfile:``, never a bare ``close()``. The
# return value decides whether the block's own exception is
# suppressed, so it must be passed through untouched.
return _run_redacting_url_credentials(
lambda: self._inner.__exit__(*exc_info))
def __getattr__(self, name: str) -> Any:
# ``_inner`` is bound in ``__init__`` and so never reaches here in
# normal use. It is named explicitly anyway: without this, an
# instance that has not run ``__init__`` -- one built by
# ``__new__``, or mid-unpickle -- turns every attribute access into
# unbounded recursion instead of an ``AttributeError``.
if name == "_inner":
raise AttributeError(name)
attr = getattr(self._inner, name)
if name in _OPTIONAL_REDACTED_OPS and callable(attr):
return _redacting_call(attr)
return attr
def _is_resource_content(content: dict) -> bool:
"""Whether an in-memory directory holds a resource's config file."""
return GR_CONF_FILE_NAME in content and \
not isinstance(content[GR_CONF_FILE_NAME], dict)
def _scan_for_resources(
content_dict: dict, parent_id: list[str],
) -> Generator[tuple[str, tuple[int, ...], dict], None, None]:
"""Yield ``(id, version, content)`` for every resource under a folder."""
for name, content in content_dict.items():
curr_id = [*parent_id, name]
curr_id_path = "/".join(curr_id)
if not isinstance(content, dict):
logger.warning(
"file <%s> is not used.",
escape_unsafe_characters(curr_id_path))
continue
try:
resource_id, version = parse_gr_id_version_token(name)
except ValueError:
logger.warning(
"skipping directory <%s> -- its name %s",
escape_unsafe_characters(curr_id_path),
malformed_resource_id_reason(name))
continue
if _is_resource_content(content):
yield "/".join([*parent_id, resource_id]), version, content
continue
if not is_gr_id_token(name):
# Parsed, so what is left is a version suffix or a separator
# -- and a folder that is not a resource can carry neither.
logger.warning(
"skipping directory <%s> -- its name is not a resource "
"id token", escape_unsafe_characters(curr_id_path))
continue
# scan children
yield from _scan_for_resources(content, curr_id)
def _scan_for_resource_files(
content_dict: dict[str, Any], parent_dirs: list[str],
) -> Generator[tuple[str, str | bytes], None, None]:
for path, content in content_dict.items():
if isinstance(content, dict):
# handle subdirectory
for fname, fcontent in _scan_for_resource_files(
content, [*parent_dirs, path]):
yield fname, fcontent
else:
fname = "/".join([*parent_dirs, path])
if isinstance(content, (str, bytes)):
# handle file content
yield fname, content
else:
logger.error(
"unexpected content at %s: %s", fname, content)
raise TypeError(f"unexpected content at {fname}: {content}")
[docs]
def build_inmemory_protocol(
proto_id: str,
root_path: str,
content: dict[str, Any]) -> FsspecReadWriteProtocol:
"""Build and return an embedded fsspec protocol for testing."""
if not os.path.isabs(root_path):
logger.error(
"for embedded resources repository we expects an "
"absolute path: %s", root_path)
raise ValueError(f"not an absolute root path: {root_path}")
proto = build_fsspec_protocol(proto_id, f"memory://{root_path}")
if not isinstance(proto, FsspecReadWriteProtocol):
# Defensive, and unreachable today -- deliberately kept anyway. This
# function's return type promises a read-write protocol, and it used
# to make that promise true by ``cast`` alone: an assertion to the
# type checker that nothing checked at run time (#528). The memory
# scheme does build read-write, so the promise held; nothing kept it
# holding.
#
# The one way the builder above can hand back something else is a memo
# hit on an id already held by a read-only protocol, and that is
# refused earlier and more informatively in ``__new__`` (#514) -- so
# this branch has no test, because no caller can reach it. It exists
# so that a future change to the builder's dispatch cannot quietly
# turn the promise false.
raise TypeError(
f"protocol {proto_id!r} over memory://{root_path} is not "
f"read-write, so it cannot hold an embedded repository")
resources: Iterable[tuple[str, tuple[int, ...], dict]]
if _is_resource_content(content):
# The repository root is itself a resource, published under the
# empty id.
resources = [("", (0,), content)]
else:
resources = _scan_for_resources(content, [])
for rid, rver, rcontent in resources:
resource = GenomicResource(rid, rver, proto)
for fname, fcontent in _scan_for_resource_files(rcontent, []):
mode = "wt"
if isinstance(fcontent, bytes):
mode = "wb"
with proto.open_raw_file(resource, fname, mode) as outfile:
outfile.write(fcontent)
proto.save_resource_file_state(
resource, proto.build_resource_file_state(resource, fname))
proto.save_manifest(resource, proto.build_manifest(resource))
return proto
#: Every keyword ``_build_filesystem`` reads, and so the whole of what a
#: keyword can configure about a protocol. **Keep in step with it** -- a
#: keyword it learns to read and this set does not is one a rebuild can go on
#: changing silently. Pinned by
#: ``test_fsspec_protocol_rebuild.py``'s drift guard.
_FILESYSTEM_KWARGS = frozenset({
"base_url", "user", "password", "endpoint_url",
})
[docs]
def canonical_public_url(public_url: str) -> str:
"""Return a public url in the one spelling two builds can be compared in.
Only for comparison -- the value a protocol reports through
``get_public_url`` stays exactly as its caller wrote it.
"""
return _display_url(public_url).rstrip("/")
def _protocol_config_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]:
"""Return the keywords that configure a protocol, ready to compare.
Only the filesystem keywords: everything else a caller passes rides along
without the protocol ever reading it -- the repository factory hands the
builder its ``cache_dir``, which configures the cache wrapped *around* the
protocol -- and so cannot make two builds over one id and url disagree.
A keyword whose value is ``None`` is dropped rather than kept, because
``_build_filesystem`` reads them all with ``.get``: an omitted keyword and
an explicit ``None`` build the identical filesystem, and the repository
factory reaches one url both ways -- a ``url``-type definition passes
neither credential keyword, an ``http``-type one passes both as ``None``.
"""
return {
key: value
for key, value in kwargs.items()
if key in _FILESYSTEM_KWARGS and value is not None
}
def _refuse_a_reconfiguring_rebuild(
cls: type[FsspecReadOnlyProtocol],
existing: FsspecRepositoryProtocol,
proto_id: str,
url: str,
kwargs: dict[str, Any],
) -> None:
"""Refuse a rebuild that asks for a differently configured protocol.
``__new__`` memoizes one instance per ``(proto_id, canonical url)`` and
never evicts, so that pair names a single protocol for the whole process.
A second build over it that asks for something *else* cannot be honoured
and used to be answered silently and wrongly instead (#514): the memo hit
was returned as-is, and everything ``__init__`` rebinds on the way out --
the public url, the credential kwargs -- was applied to the instance every
existing holder was already using.
Mode is the sharpest case, because ``FsspecReadWriteProtocol`` subclasses
the read-only protocol. A read-only build over a read-write key satisfied
``isinstance`` and got a *writable* protocol with the read-write
``__init__`` re-run on it, while a read-write build over a read-only key
got an instance Python did not call ``__init__`` on at all -- stale
filesystem, stale kwargs, and no write methods to fail on until much
later.
Reported here, where the wrong thing is asked for, rather than as an absent
write method or an unexpected url somewhere downstream. Two genuinely
different protocols over one url are still available -- under two ids.
"""
requested = (
Mode.READWRITE
if issubclass(cls, ReadWriteRepositoryProtocol)
else Mode.READONLY
)
if existing.mode() != requested:
raise ValueError(
f"protocol {proto_id!r} over {strip_url_userinfo(url)} is "
f"already built as {existing.mode().name}; it cannot also serve "
f"a {requested.name} build -- give the {requested.name} protocol "
f"an id of its own")
# The incumbent's attributes are always there to compare against:
# ``_FSSPEC_PROTOCOLS`` holds only protocols whose whole construction has
# returned (#527). This used to need an early return for an incumbent
# published before ``__init__`` had configured it -- reading
# ``public_url`` anyway turned a race that merely ran ``__init__`` twice
# into an ``AttributeError`` raised out of ``__new__``. That tolerance is
# deliberately NOT kept now that the window is closed: it would let a
# regression that reopened it answer a divergent rebuild silently instead
# of being found. See ADR 0005.
requested_public_url = kwargs.get("public_url")
if requested_public_url is None:
requested_public_url = _display_url(url)
# Compared in one spelling, not as authored. An incumbent's ``public_url``
# is whatever its caller passed, while a rebuild that passes none defaults
# to the url's display form -- so a trailing slash, or a bare path against
# its ``file://`` form, would otherwise read as a request to republish the
# repository somewhere else.
if canonical_public_url(requested_public_url) != \
canonical_public_url(existing.public_url):
raise ValueError(
f"protocol {proto_id!r} over {strip_url_userinfo(url)} is "
f"already built with the public url "
f"{strip_url_userinfo(existing.public_url)}; rebuilding it "
f"cannot repoint it at "
f"{strip_url_userinfo(requested_public_url)} -- give the "
f"protocol published under that url an id of its own")
requested_kwargs = _protocol_config_kwargs(kwargs)
existing_kwargs = _protocol_config_kwargs(existing.kwargs)
if requested_kwargs != existing_kwargs:
disagreeing = sorted(
(set(requested_kwargs) ^ set(existing_kwargs)) | {
key
for key in set(requested_kwargs) & set(existing_kwargs)
if requested_kwargs[key] != existing_kwargs[key]
})
# The disagreeing KEYS, never their values: these keywords are how
# http basic-auth credentials reach a protocol, and an exception
# message is logged, echoed and reported.
raise ValueError(
f"protocol {proto_id!r} over {strip_url_userinfo(url)} is "
f"already built with a different {', '.join(disagreeing)}; "
f"rebuilding it cannot reconfigure the protocol its holders are "
f"using -- give the differently configured protocol an id of "
f"its own")
class _ProtocolConstruction:
"""One in-flight construction of a memoized protocol key.
A protocol becomes reachable through ``_FSSPEC_PROTOCOLS`` only once it is
configured, so between ``__new__`` creating the instance and ``__init__``
(or ``__setstate__``) finishing with it, the construction needs somewhere
else to be recorded -- otherwise a second thread over the same key cannot
tell "nobody has built this" from "somebody is building it", and builds a
second instance over a key that is supposed to name one (#527).
It is released exactly once, whether the construction was published or
abandoned, so a waiter is never left blocked by a build that raised.
The instance is held WEAKLY, and that is load-bearing rather than tidy:
the two guards that release a key -- ``_BuiltOnceProtocolMeta.__call__``
and ``__setstate__`` -- both sit *after* ``__new__`` has taken it, and the
deserialize path can die in between, when a frame carrying a protocol is
truncated or corrupt, or when a ``KeyboardInterrupt`` lands there. Neither
guard is reached then, so nothing announces the abandonment. What is left
to go on is that the half-built instance itself is dropped: a construction
whose instance has been collected is one nobody can ever finish, and
``__new__`` takes such a key on rather than waiting for it.
"""
def __init__(
self, instance: FsspecReadOnlyProtocol, key: tuple[str, str],
) -> None:
self.key = key
self._instance = ref(instance)
self._done = Event()
self._owner = get_ident()
self._finalizer = finalize(instance, self._instance_collected)
# Process teardown is not an abandonment worth reporting, and a
# construction still in flight then has no waiter left to wake.
self._finalizer.atexit = False
@property
def instance(self) -> FsspecReadOnlyProtocol | None:
"""The instance being constructed, or ``None`` once it is dropped."""
return self._instance()
def _instance_collected(self) -> None:
"""Wake the waiters of a construction that can no longer finish.
Deliberately takes no lock. A finalizer runs at whatever allocation
point collects the instance -- including one inside
``_FSSPEC_PROTOCOLS_GUARD`` -- so acquiring that guard here could
deadlock the very thread it interrupted. Waking is enough: the record
is dropped by ``__new__``, under the guard, the next time anything
looks at the key.
"""
self._done.set()
def reentered_by_this_thread(self) -> FsspecReadOnlyProtocol | None:
"""The instance, if this thread is the one constructing the key.
Only read to keep a re-entrant construction from waiting on itself.
Nothing in the tree builds a protocol from inside a protocol's
constructor, so this guards against a deadlock rather than describing
a supported shape. The answer cannot be a live construction with a
collected instance -- the thread inside it is holding the instance.
"""
if self._owner != get_ident():
return None
return self._instance()
def wait(self) -> None:
"""Block until this construction is published, abandoned or dropped."""
self._done.wait()
def release(self) -> None:
"""Wake everything waiting on this construction."""
# Detached rather than left armed: a published protocol is never
# evicted, so its finalizer would hold this record -- and the
# construction bookkeeping behind it -- for the life of the process.
self._finalizer.detach()
self._done.set()
def _finish_construction(instance: FsspecReadOnlyProtocol) -> None:
"""Publish a protocol that has finished being configured.
A no-op unless this instance is the one under construction for its key: a
rebuild of a memoized protocol re-runs ``__init__`` on the live instance,
and that refresh is not a publication.
"""
construction = instance.__dict__.pop("_construction", None)
if construction is None:
return
with _FSSPEC_PROTOCOLS_GUARD:
if _FSSPEC_PROTOCOLS_UNDER_CONSTRUCTION.get(
construction.key) is construction:
del _FSSPEC_PROTOCOLS_UNDER_CONSTRUCTION[construction.key]
_FSSPEC_PROTOCOLS[construction.key] = instance
construction.release()
def _abandon_construction(instance: FsspecReadOnlyProtocol) -> None:
"""Drop a construction that raised, leaving its key buildable again.
The alternative -- leaving the record in place -- would make one failed
build permanently unbuildable AND block every thread already waiting on
it forever, which is a far worse outcome than the duplicated ``__init__``
the serialisation replaced.
"""
construction = instance.__dict__.pop("_construction", None)
if construction is None:
return
with _FSSPEC_PROTOCOLS_GUARD:
if _FSSPEC_PROTOCOLS_UNDER_CONSTRUCTION.get(
construction.key) is construction:
del _FSSPEC_PROTOCOLS_UNDER_CONSTRUCTION[construction.key]
construction.release()
class _BuiltOnceProtocolMeta(abc.ABCMeta):
"""Publishes a protocol only once its whole construction has returned.
``__new__`` cannot publish the instance itself, because Python runs
``__init__`` only after ``__new__`` has returned -- which is exactly the
window this class closes (#527). Nor can the base ``__init__``:
``FsspecReadWriteProtocol.__init__`` creates the repository root *after*
the base constructor body, so a protocol published from there would be
handed to a waiting thread before its root existed.
Taking over the call protocol gives one place that is after the whole
``__init__`` chain and before the caller receives the object. It is also
the only place that sees a construction failing *before* ``__init__`` is
entered -- a call-signature error -- which would otherwise strand the key
and block every later builder of it forever.
Unpickling does not come through here at all: it runs
``__getnewargs_ex__`` -> ``__new__`` -> ``__setstate__`` and never calls
``__init__``, so ``__setstate__`` finishes its own construction.
"""
def __call__(cls, *args: Any, **kwargs: Any) -> Any:
# ``cls`` is the protocol class being instantiated, not the metaclass;
# the cast says so, since ``ABCMeta.__new__`` is a class-creation
# signature and this is the instance-creation one.
target = cast("type[Any]", cls)
instance = target.__new__(target, *args, **kwargs)
if not isinstance(instance, cls):
# ``type.__call__``'s own rule, kept: an instance of another class
# is not initialised. Unreachable here -- a memo hit of another
# class is a mode mismatch, and refused in ``__new__``.
return instance
try:
# pylint: disable=unnecessary-dunder-call
type(instance).__init__(instance, *args, **kwargs)
except BaseException:
_abandon_construction(instance)
raise
_finish_construction(instance)
return instance
[docs]
class FsspecReadOnlyProtocol(
ReadOnlyRepositoryProtocol, metaclass=_BuiltOnceProtocolMeta):
"""Provides fsspec genomic resources repository protocol.
``(proto_id, url)`` names ONE protocol instance for the life of the
process: ``__new__`` memoizes it in ``_FSSPEC_PROTOCOLS``, keyed on the
url's canonical ``scheme://netloc/path`` form so its spelling cannot split
one repository in two, and never evicts. A second build over that pair
therefore reaches the object every earlier caller is already holding, and
Python re-runs ``__init__`` on it.
That makes a rebuild a *refresh* -- it drops the resource memo, which is
how a caller that has just changed a repository reads it back. It is not a
reconfiguration: a rebuild asking for a different mode, public url or
credentials is refused rather than applied to the incumbent. See
``docs/adr/0005-fsspec-protocol-memo-rebuild.md`` (#514).
Construction is one atomic step, and a protocol is reachable through the
memo only once the whole of it has returned (#527). ``__new__`` records an
in-flight construction instead of publishing, so a thread that arrives
while another is building that key waits and is answered with the same,
configured instance -- rather than either building a second protocol over
a key that names one, or reading one whose ``filesystem``, ``url``,
``public_url`` and ``kwargs`` are not bound yet.
"""
#: The repository's resources, memoized on first read, and the lock that
#: guards it. Bound once per instance in ``__new__`` rather than in
#: ``__init__``, because ``__init__`` re-runs on every memoized instance a
#: rebuild hands back and must not replace the lock its readers are
#: already holding (#514).
_all_resources: dict[str, GenomicResource] | None
_all_resources_lock: Lock
#: Present only while this instance is being constructed, and removed by
#: whichever of ``_finish_construction``/``_abandon_construction`` gets to
#: it -- so its presence is what tells a rebuild's ``__init__`` apart from
#: the first one (#527).
_construction: _ProtocolConstruction
def __getnewargs_ex__(self) -> tuple[tuple, dict]:
# pylint: disable=invalid-getnewargs-ex-returned
# self.kwargs may hold HTTP basic-auth credentials (user/password).
# They are INTENTIONALLY pickled with the protocol so a dask worker
# deserializing this protocol can rebuild an authenticated
# filesystem and read the remote GRR. Do not strip them here — that
# would break distributed reads of an authed http repository.
# Pickle the credential-bearing fetch url (not the stripped display
# url) so a dask worker rebuilds an authenticated protocol whose cache
# key matches a fresh build, and ``__init__`` (were it called) would
# re-derive the same stripped ``self.url``. The credential
# re-materializes here on purpose — see the class docstring above.
args = (self.proto_id, self._fetch_url)
kwargs: dict[str, Any] = copy.copy(self.kwargs)
kwargs["public_url"] = self.public_url
return (args, kwargs)
def __new__(cls, *args: Any, **kwargs: Any) -> FsspecReadOnlyProtocol:
proto_id = args[0] if len(args) > 0 else kwargs["proto_id"]
url = args[1] if len(args) > 1 else kwargs["url"]
# The cache KEY is kept credentialed on purpose: keying on the
# userinfo-stripped url would let a second build with DIFFERENT
# credentials for the same host+path reuse the first protocol and
# authenticate with the WRONG credentials. The ``_FSSPEC_PROTOCOLS``
# dict/key is never logged, repr'd or serialized, so retaining the
# credential in the key does not leak it. The DEBUG line below, which IS
# a leak vector, is passed a userinfo-stripped url. For a userinfo-free
# url the stripped url == url, so behavior is unchanged.
#
# Canonicalised, so the spelling of the url cannot split one repository
# across two entries: several builders pass a bare ``/abs/path`` while
# ``__getnewargs_ex__`` pickles the ``file://`` form, and the two used
# to be different keys -- which is how a pickle round trip minted a
# second protocol over one directory (#514).
key = (proto_id, _fetch_url_form(url))
# The memo read, the instance creation and the record of an in-flight
# construction are one step, under one lock. They used to be a plain
# check-then-set: two threads that both missed for one key both built,
# and the second publication replaced the first, so the loser walked
# away holding an orphan protocol the memo does not know about -- with
# a resource memo and a lock of its own, outside the mutual exclusion
# of #458 and outside the rebuild refusal of #514 (#527).
#
# The instance is NOT published here. ``__init__`` runs only after
# ``__new__`` has returned, so anything published from here is
# reachable with ``filesystem``, ``url``, ``public_url`` and
# ``kwargs`` still unbound; publication is
# ``_finish_construction``'s, driven by ``_BuiltOnceProtocolMeta``
# for a normal build and by ``__setstate__`` for an unpickle.
while True:
with _FSSPEC_PROTOCOLS_GUARD:
existing = _FSSPEC_PROTOCOLS.get(key)
if existing is not None:
_refuse_a_reconfiguring_rebuild(
cls, existing, proto_id, url, kwargs)
logger.debug(
"protocol with id %s and url %s already exists, "
"returning the existing instance",
proto_id, strip_url_userinfo(url))
return existing
pending = _FSSPEC_PROTOCOLS_UNDER_CONSTRUCTION.get(key)
if pending is not None:
reentered = pending.reentered_by_this_thread()
if reentered is not None:
# This thread is already constructing this key.
# Nothing in the tree does that, and waiting would be
# waiting on ourselves.
return reentered
# Asked without binding the answer to a local: a strong
# reference held across the wait below would pin the
# half-built instance, and its collection is the whole
# signal being read here.
if pending.instance is None:
# Nobody holds the half-built instance any more, so
# nobody is coming back to finish it: the deserialize
# that took this key died between ``__new__`` and
# ``__setstate__``, where neither of the guards that
# release a key is reached. Drop the record and take
# the key on -- leaving it would block every builder
# of it for the life of the process, in an untimed
# wait and with no log line (#527).
del _FSSPEC_PROTOCOLS_UNDER_CONSTRUCTION[key]
pending = None
if pending is None:
instance = super().__new__(cls)
# Before the instance is reachable, and never again: the
# memo lock is the one piece of state a rebuild must not
# touch, so it is bound where construction happens
# exactly once (#514).
instance._all_resources_lock = Lock()
instance._all_resources = None
construction = _ProtocolConstruction(instance, key)
_FSSPEC_PROTOCOLS_UNDER_CONSTRUCTION[key] = construction
instance._construction = construction
return instance
# Waited for OUTSIDE the guard: a construction runs the whole
# ``__init__`` chain, which for a read-write protocol does
# filesystem I/O, and holding the memo lock across that would put
# every protocol build in the process behind one remote round
# trip. Then round the loop: the key is either published by now,
# or the construction was abandoned and this thread takes it on.
pending.wait()
def __init__(
self, proto_id: str,
url: str, *,
filesystem: fsspec.AbstractFileSystem,
public_url: str | None = None,
**kwargs: Any,
):
super().__init__(proto_id, url)
parsed = urlparse(url)
self.scheme = parsed.scheme
if self.scheme == "":
self.scheme = "file"
# ``self.netloc``/``self.url``/``self.public_url`` are the DISPLAY /
# IDENTITY of this protocol — returned by ``get_url``/``get_public_url``
# and serialized into web responses, persisted docs and logs. They MUST
# NOT carry credentials, so any ``user:pass@`` userinfo embedded in a
# ``scheme://user:pass@host`` url is stripped from them here.
fetch_netloc = parsed.netloc
self.netloc = _strip_netloc_userinfo(fetch_netloc)
self.root_path = parsed.path
self.url = _display_url(url)
# ``self._fetch_url`` is the credential-BEARING url used only to talk to
# the remote filesystem. For URL-embedded userinfo, aiohttp/htslib read
# the basic-auth credentials straight from this url string (they are not
# in ``kwargs``), so every fetched file url must derive from it — see
# ``get_resource_url``/``load_contents``/``md5_contents``. When the url
# has no userinfo this is byte-identical to ``self.url``. It is also
# what the memo is keyed on, so it and the key cannot drift.
self._fetch_url = _fetch_url_form(url)
if public_url is None:
self.public_url = self.url
else:
self.public_url = public_url
self.filesystem = filesystem
# kwargs may carry HTTP basic-auth credentials (user/password). They
# are kept so the filesystem can be rebuilt after unpickling on a
# dask worker (see __getnewargs_ex__/__setstate__); they are never
# logged and are masked in the definition model's repr.
self.kwargs: dict[str, Any] = kwargs
# This body re-runs on the LIVE instance whenever a memoized protocol
# is rebuilt (see ``__new__``), so a rebuild is a refresh of the
# resource memo -- ``grr_manage`` re-reading a repository it has just
# changed depends on that. Take the incumbent's own lock to do it,
# which is what ``invalidate`` is; rebinding a fresh ``Lock`` here and
# clearing the memo beside it left a reader inside the guard with no
# mutual exclusion at all (#514).
#
# The other assignments above are rebound on the live instance too, and
# are safe for narrower reasons: the url fields are derived from the
# memo key, so they cannot differ; ``public_url`` and the filesystem
# keywords are what ``__new__`` refuses a rebuild over; ``filesystem``
# is a freshly built but equivalent object, because every production
# construction in the tree routes through ``build_fsspec_protocol``,
# which derives it from the key and those keywords.
#
# One test helper is deliberately outside that:
# ``testing.build_faulty_test_protocol`` hands in a scripted
# filesystem of its own, for which a rebind would NOT be equivalent
# -- it would give the incumbent's holders a filesystem scripted by
# somebody else. Nothing here can refuse that, because the rebuild
# guard compares the credential keywords and the public url and a
# filesystem instance has no value equality to compare; the helper
# closes it instead, by refusing to build twice over one root. See
# ADR 0021 and #874. ``kwargs`` as a whole is NOT
# equal -- a keyword the protocol never reads (the factory's
# ``cache_dir``) may differ, and the second caller's value wins.
# Nothing reads those, and rebinding a reference is atomic, so a
# concurrent reader cannot catch any of it half-done.
self.invalidate()
def __getstate__(self) -> dict[str, Any]:
state = self.__dict__.copy()
# Remove the unpicklable entries.
del state["filesystem"]
del state["_all_resources"]
del state["_all_resources_lock"]
# Defensive: an in-flight construction is this process's bookkeeping
# (and holds an ``Event``, which does not pickle). A configured
# protocol has none, so this only ever fires for a protocol pickled
# from inside its own constructor.
state.pop("_construction", None)
return state
def __setstate__(self, state: dict[str, Any]) -> None:
# Unpickling is the one path that reaches ``__new__`` without
# ``__init__`` ever running -- ``__getnewargs_ex__`` -> ``__new__`` ->
# here -- so this is where a deserialize into a COLD memo finishes its
# own construction. A scheme that published from ``__init__`` alone
# would leave the key in flight forever and hang every later builder
# of it (#527).
try:
self.__dict__.update(state)
self.filesystem = _build_filesystem(
self._fetch_url, **self.kwargs)
# In a process that already has this protocol, ``__new__`` landed
# on the live instance, so this is the same rebuild ``__init__``
# performs -- and the same reason not to rebind the lock a reader
# is holding here either (#514).
self.invalidate()
except BaseException:
_abandon_construction(self)
raise
_finish_construction(self)
[docs]
def get_url(self) -> str:
return self.url
[docs]
def get_resource_url(self, resource: GenomicResource) -> str:
# Fetch path: derive resource file urls from the credential-bearing
# ``_fetch_url`` (the base class uses ``self.url``, which is stripped of
# userinfo for display) so URL-embedded basic-auth still reaches
# aiohttp/htslib. Identical to the base for userinfo-free urls.
#
# A join of its own, so it repeats the id containment check the base
# class does -- this override is the one that serves the REMOTE
# repositories, whose ids come out of an untrusted ``.CONTENTS``
# (gain#467).
validate_resource_id(resource.resource_id)
return os.path.join(
self._fetch_url,
resource.get_full_id())
[docs]
def get_public_url(self) -> str:
return self.public_url
[docs]
def invalidate(self) -> None:
"""Drop the memoized resources, leaving handed-out ones alone.
Clears only this protocol's own cache. The resources in the memo are
handed out by reference, so unbinding their ``proto`` on the way out
-- as this used to do -- breaks the objects live callers are already
holding, and they raise ``AttributeError`` on ``None`` at first use
(#513). Their lifetime is the caller's business; dropping the memo is
enough to make the next read reload, and enough to let a resource no
one else holds be collected, since the memo held the only reference
to it. Not the protocol, though -- ``_FSSPEC_PROTOCOLS`` memoizes
every protocol for the life of the process and never evicts, so no
amount of unbinding here ever released one.
"""
# Under the memo lock, and over the whole body: returning the memo
# from inside the lock is only half the guarantee, because an
# unsynchronised ``invalidate`` can still clear the attribute in the
# middle of a populating read (#458).
with self._all_resources_lock:
self._all_resources = None
[docs]
def close(self) -> None:
"""Close the genomic resource."""
self.invalidate()
def _read_fetch_file(
self, filepath: str, mode: str, compression: str | None,
) -> str | bytes:
"""Open+read a fetch-url file, redacting any credential on failure.
``_run_redacting_url_credentials``'s guarantee, covering both the
open and the read.
"""
def open_and_read() -> str | bytes:
with self.filesystem.open(
filepath, mode, compression=compression) as infile:
return cast("str | bytes", infile.read())
return _run_redacting_url_credentials(open_and_read)
[docs]
def load_contents(self) -> list[dict[str, Any]]:
"""Load the content JSON of the repository."""
content_filename = os.path.join(
self._fetch_url, GR_CONTENTS_FILE_NAME)
compression: str | None = "gzip"
if not self.filesystem.exists(content_filename):
# Not a dead branch: nothing has written this since #758,
# but repositories an older release published still carry
# one and no other file describes them.
content_filename = os.path.join(
self._fetch_url, GR_LEGACY_CONTENTS_FILE_NAME)
compression = None
data = self._read_fetch_file(content_filename, "rt", compression)
return cast(list[dict[str, Any]], json.loads(data))
[docs]
def md5_contents(self) -> str:
"""Calculate md5 hash of the repository content."""
content_filename = os.path.join(
self._fetch_url, GR_CONTENTS_FILE_NAME)
if not self.filesystem.exists(content_filename):
# See `load_contents`: the legacy index is still the only
# description an older repository has.
content_filename = os.path.join(
self._fetch_url, GR_LEGACY_CONTENTS_FILE_NAME)
data = self._read_fetch_file(content_filename, "rb", None)
assert isinstance(data, bytes)
return hashlib.md5(data).hexdigest() # ruff: ignore[hashlib-insecure-hash-function]
[docs]
def get_all_resources(self) -> Generator[GenomicResource, None, None]:
"""Return generator over all resources in the repository."""
yield from self.get_all_resources_dict().values()
def _enumerate_resources(self) -> Iterable[GenomicResource]:
"""Enumerate this repository's resources, in any order.
The one seam ``get_all_resources_dict`` leaves to a subclass: that
method holds the memo, the lock, the keying and the ordering, so
there is one implementation of the memo protocol to get right rather
than one per protocol (#515).
Called with ``_all_resources_lock`` HELD, and the lock is NOT
reentrant. An implementation must therefore not take it, and must not
re-enter ``get_all_resources_dict``, ``get_all_resources``, ``close``
or ``invalidate`` -- on this protocol or on any protocol whose
invalidation cascades back to this one. That is a real trap and not a
theoretical one: ``CachingProtocol.get_all_resources_dict``
legitimately invalidates a *sub*-protocol from inside its own memo
population, and the same idiom here deadlocks the process with no
traceback. Pinned by
``test_the_seam_runs_with_the_memo_lock_held``.
Named apart from the public ``collect_all_resources`` on purpose:
that one is the read-write scan itself, and this one is the memo's
seam. Overriding the wrong one is silent.
This implementation reads the repository's ``.CONTENTS``, the only
enumeration available to a protocol that cannot scan for itself.
"""
all_resources = []
contents = self.load_contents()
for entry in contents:
# ``.CONTENTS`` is remote, untrusted GRR content and its ``id``
# is joined onto the repository url, so a traversing id reads --
# and, through the caching repository, WRITES -- outside the
# root. Dropped with a warning rather than raised on: one
# poisoned entry must not cost the repository its healthy
# resources (gain#467).
# Malformedness is the second, separate rule, dropped the
# same way: an id a scan would have refused must not enter
# here either (gain#1352).
resource_id = entry["id"]
reason = (
uncontained_resource_id_reason(resource_id)
or malformed_resource_id_reason(resource_id)
)
if reason is not None:
logger.warning(
"repo %s: dropping resource <%s> from %s -- "
"its id %s",
self.proto_id, escape_unsafe_characters(resource_id),
GR_CONTENTS_FILE_NAME, reason)
continue
version = tuple(map(int, entry["version"].split(".")))
manifest = Manifest.from_manifest_entries(entry["manifest"])
resource = self.build_genomic_resource(
resource_id, version, config=entry["config"],
manifest=manifest)
logger.debug(
"repo %s loaded resource %s",
self.proto_id,
resource.resource_id)
all_resources.append(resource)
return all_resources
[docs]
def get_all_resources_dict(self) -> dict[str, GenomicResource]:
"""Return the repository's resources, keyed by full id.
The whole memo protocol -- the lock, the check-then-populate, the
keying, the ordering and the return -- lives here and only here, for
every fsspec protocol. A subclass enumerates the repository by
overriding ``_enumerate_resources`` and inherits the rest (#515).
``FsspecReadWriteProtocol`` used to carry a second copy of all of it
and differ only in the enumeration, which is how the release-then-read
defect of #458 came to be in two places while the report named one.
"""
with self._all_resources_lock:
if self._all_resources is None:
# Ordered here rather than in the seam: the memo's key order
# is this method's guarantee, so an enumeration cannot cost
# the repository its ordering by yielding as it finds.
#
# ``sorted`` drains the seam's iterable HERE, inside the
# lock. A seam that returns a generator therefore still does
# all of its work under the lock, exactly as the two
# hand-written copies did.
self._all_resources = {
res.get_full_id(): res
for res in sorted(
self._enumerate_resources(),
key=lambda r: r.get_full_id(),
)
}
# Returned from inside the lock: reading ``self._all_resources``
# once the lock has been released is a second, unsynchronised
# read of the attribute, and an ``invalidate`` landing between
# the two hands the caller ``None`` (#458).
return self._all_resources
[docs]
def file_exists(
self, resource: GenomicResource, filename: str) -> bool:
filepath = self.get_resource_file_url(resource, filename)
return cast(bool, self.filesystem.exists(filepath))
[docs]
def load_manifest(self, resource: GenomicResource) -> Manifest:
"""Load resource manifest."""
content = self.get_file_content(resource, GR_MANIFEST_FILE_NAME)
return Manifest.from_file_content(content)
[docs]
def get_file_content(
self, resource: GenomicResource,
filename: str,
*,
uncompress: bool = True, # ruff: ignore[unused-method-argument]
mode: str = "t",
) -> Any:
"""Return content of a file in given resource.
Overrides the base, which opens through ``open_raw_file`` and reads
on the handle it returns: that open is redacted and the read is not,
so on an authed GRR a failure mid-read surfaces the credential-bearing
fetch url verbatim (gain#1058). ``_read_fetch_file`` covers both.
This sits IN FRONT of the fasta-index copy gain#1017 fixed --
``ReferenceGenome.open`` reads the ``.fai`` through here first -- and
also backs ``load_manifest`` and ``load_yaml``.
``get_loaded_manifest`` reads a missing manifest's
``FileNotFoundError`` as "no manifest", and the rebuild keeps that
type: it reconstructs via ``type(exc)(message)``, which
``FileNotFoundError`` supports.
An error that CANNOT be reconstructed from a message alone loses its
type -- an ``aiohttp.ClientResponseError`` (an HTTP 5xx, which fsspec
does not translate) needs ``request_info`` and ``history``. It comes
back an ``OSError``, or a ``RetryableCopyError`` where the original
was transient, which ``_rebuild_error_without_url_credentials``
preserves so redaction cannot change a retry decision. Safe here for the
reason ``_copy_resource_file_to_local`` sets out: no retry or
control-flow decision on this path keys off the type.
``uncompress`` names nothing on this path and never has, so
``compression=None`` preserves the as-stored read exactly -- see
``_copy_resource_file_to_local``, which found the same. Repairing the
dead parameter is separate work.
"""
filepath = self.get_resource_file_url(resource, filename)
return self._read_fetch_file(filepath, f"r{mode}", None)
[docs]
def open_raw_file(
self, resource: GenomicResource, filename: str,
mode: str = "rt", **kwargs: str | bool | None) -> IO:
filepath = self.get_resource_file_url(resource, filename)
if "w" in mode:
if self.mode() == Mode.READONLY:
# ``filepath`` derives from the credential-BEARING
# ``_fetch_url``, and this message is interpolated before any
# handle exists, so the ``_RedactingFile`` wrapper of ADR 0023
# cannot reach it. Redacted here by hand, as the download
# loop's "destination file not created" path already is
# (gain#620, gain#1106). It matters more than a propagated
# error: ``OSError`` is in ``RESOURCE_ERRORS``, so
# ``report_resource_failure`` logs this text at ERROR.
raise OSError(
f"Read-Only protocol {self.get_id()} trying to open "
f"{strip_url_userinfo(filepath)} for writing")
# Create the containing directory if it doesn't exists.
parent = os.path.dirname(filepath)
if not self.filesystem.exists(parent):
self.filesystem.mkdir(
parent, create_parents=True, exist_ok=True)
compression = None
if kwargs.get("compression"):
compression = "gzip"
return self._open_fsspec_file(filepath, mode, compression)
def _open_fsspec_file(
self, filepath: str, mode: str,
compression: str | None) -> IO:
"""Open ``filepath`` on the filesystem, redacting a failing url.
``_run_redacting_url_credentials``'s guarantee, covering the open AND
every subsequent operation on the returned handle: it comes back
wrapped in a ``_RedactingFile`` (gain#1078). A caller that opens and
reads in one place may still use ``_read_fetch_file``, which says so
in one call.
"""
opened = _run_redacting_url_credentials(lambda: cast(
IO,
self.filesystem.open(
filepath, mode=mode,
compression=compression)))
return cast(IO, _RedactingFile(opened))
def _get_file_url(self, resource: GenomicResource, filename: str) -> str:
def process_file_url(url: str) -> str:
if self.scheme == "file":
return urlparse(url).path
if self.scheme == "s3":
return cast(str, self.filesystem.sign(
url, expiration=S3_PRESIGN_EXPIRATION_SECONDS))
return url
return process_file_url(self.get_resource_file_url(resource, filename))
[docs]
def open_tabix_file(
self, resource: GenomicResource,
filename: str,
index_filename: str | None = None) -> pysam.TabixFile:
if self.scheme not in {"file", "s3", "http", "https"}:
raise OSError(
f"tabix files are not supported on schema {self.scheme}")
file_url = self._get_file_url(resource, filename)
if index_filename is None:
# The index may be a ``.tbi`` or a ``.csi``; ask the manifest
# which one this resource actually carries (gain#430).
index_filename = resolve_tabix_index_filename_for_read(
resource, filename)
index_url = self._get_file_url(resource, index_filename)
return _open_htslib_file(
file_url,
lambda: pysam.TabixFile( # pylint: disable=no-member
file_url, index=index_url, encoding="utf-8",
parser=pysam.asTuple()))
[docs]
def open_vcf_file(
self, resource: GenomicResource,
filename: str,
index_filename: str | None = None) -> pysam.VariantFile:
if self.scheme not in {"file", "s3", "http", "https"}:
raise OSError(
f"vcf files are not supported on schema {self.scheme}")
file_url = self._get_file_url(resource, filename)
if index_filename is not None:
# Asked for BY NAME -- a table's ``index_filename``, or the
# index the caching protocol just refreshed. Refuse a name that
# names nothing rather than dropping it and opening unindexed:
# htslib would then auto-probe its way to the adjacent index and
# read on, and a configuration that does nothing would stay
# invisible (gain#596).
if not resource.file_exists(index_filename):
raise OSError(
f"index '{index_filename}' of '{filename}' not found in "
f"resource '{resource.resource_id}'")
else:
# The index may be a ``.tbi`` or a ``.csi``; ask the manifest
# which one this resource actually carries, exactly as
# ``open_tabix_file`` above does -- do NOT assume ``.tbi``
# (gain#430, gain#596).
index_filename = resolve_tabix_index_filename_for_read(
resource, filename)
if not resource.file_exists(index_filename):
# Nothing resolved: a file that ships no index at all still
# opens, unindexed (gain#596).
return _open_htslib_file(
file_url,
lambda: pysam.VariantFile( # pylint: disable=no-member
file_url))
index_url = self._get_file_url(resource, index_filename)
vcf_file = _open_htslib_file(
file_url,
lambda: pysam.VariantFile( # pylint: disable=no-member
file_url, index_filename=index_url))
_declare_index_contigs(vcf_file)
return vcf_file
[docs]
def open_fasta_file(
self, resource: GenomicResource,
filename: str,
index_filename: str | None = None,
compressed_index_filename: str | None = None) -> pysam.FastaFile:
if self.scheme not in {"file", "s3", "http", "https"}:
raise OSError(
f"fasta files are not supported on schema {self.scheme}")
if index_filename is None:
index_filename = f"{filename}.fai"
if compressed_index_filename is None:
compressed_index_filename = f"{filename}.gzi"
if not self.file_exists(resource, compressed_index_filename):
raise ValueError(
f"bgzip index '{compressed_index_filename}' is required to "
f"read bgzipped genome '{filename}' in resource "
f"'{resource.resource_id}'; generate the .fai and .gzi "
f"indexes with 'samtools faidx {filename}'")
file_url = self._get_file_url(resource, filename)
if self.scheme == "file":
return pysam.FastaFile( # pylint: disable=no-member
file_url,
filepath_index=self._get_file_url(resource, index_filename),
filepath_index_compressed=self._get_file_url(
resource, compressed_index_filename))
# Remote scheme: pysam.FastaFile requires the index arguments to be
# local paths (it os.path.exists-checks them), but htslib can range-
# read the data file remotely. Copy the small .fai/.gzi indexes to a
# temporary local directory and open against those; htslib loads both
# indexes into memory at open, so the temp files can be removed
# immediately afterwards. The multi-GB data file stays remote.
with tempfile.TemporaryDirectory(prefix="gain-fasta-idx-") as tmpdir:
local_index = self._copy_resource_file_to_local(
resource, index_filename, tmpdir)
local_compressed_index = self._copy_resource_file_to_local(
resource, compressed_index_filename, tmpdir)
# Only this branch needs the guard: ``file_url`` is the one url
# that stays remote and reaches htslib, and the ``file`` branch
# above hands pysam a bare filesystem path, which can carry
# neither userinfo nor a signed query string.
return _open_htslib_file(
file_url,
lambda: pysam.FastaFile( # pylint: disable=no-member
file_url,
filepath_index=local_index,
filepath_index_compressed=local_compressed_index))
def _copy_resource_file_to_local(
self, resource: GenomicResource,
filename: str, dest_dir: str) -> str:
"""Copy a (small) resource file into dest_dir; return the local path.
Reads through ``_read_fetch_file`` rather than ``open_raw_file``:
on an authed GRR a failure mid-read carries the credential-bearing
fetch url in its own message (gain#1017). ``_open_fsspec_file``
states why the open-level redaction cannot reach it.
Redacting at the read is safe here because no retry or control-flow
decision on this path keys off the exception type -- unlike
``copy_resource_file``, whose retry loop does. That loop is no longer
a reason to redact late anywhere: since gain#1078 the rebuild
preserves retryability (see ``_rebuild_error_without_url_credentials``).
One type test is reachable, ``report_resource_failure``'s
``RESOURCE_ERRORS`` check, and the rebuild helps there rather than
hurting: an ``aiohttp`` error is not an ``OSError``, so it takes
the "unexpected internal error" branch that logs the whole
credential-bearing chain with ``exc_info``, while the rebuilt
``OSError`` takes the redacted-message branch.
"""
dest = os.path.join(dest_dir, os.path.basename(filename))
filepath = self.get_resource_file_url(resource, filename)
# No compression: the old ``uncompress=False`` named nothing --
# ``open_raw_file`` reads ``compression``, never ``uncompress`` --
# so the file was always read as stored, and still is.
data = self._read_fetch_file(filepath, "rb", None)
assert isinstance(data, bytes)
pathlib.Path(dest).write_bytes(data)
return dest
[docs]
def open_bigwig_file(
self, resource: GenomicResource, filename: str,
) -> Any:
"""Open ``filename`` of ``resource`` with pyBigWig.
A ``file`` GRR is opened by path. An ``s3``, ``http`` or ``https``
GRR is opened by url -- presigned for ``s3`` -- which needs a
pyBigWig built with libcurl (``pyBigWig.remote == 1``); the PyPI
wheel is not, and a remote open on it is refused with an ``OSError``
naming the remedies. That refusal never carries the url, which may
hold a credential.
"""
if self.scheme not in {"file", "s3", "http", "https"}:
raise OSError(
f"bigwig files are not supported on schema {self.scheme}")
has_remote = pyBigWig.remote # pylint: disable=I1101
if self.scheme != "file" and not has_remote:
raise OSError(
f"cannot open bigwig file {filename} of resource "
f"{resource.resource_id} over {self.scheme}: this pyBigWig "
"build has no remote-file support (pyBigWig.remote == 0); "
"give the repository a cache_dir so the file is fetched "
"locally, or install a curl-enabled pyBigWig (e.g. the "
"bioconda build)")
file_url = self._get_file_url(resource, filename)
# Not the verbosity bracket of ``_open_htslib_file``: libBigWig is not
# htslib and ``pysam.set_verbosity`` does not reach it. Its
# ``[urlOpen]`` line goes to fd 2 through its own ``fprintf``, so that
# descriptor is what has to be taken away (gain#1333). The redaction
# rides along: pyBigWig's own exception carries no url today, which
# makes that half a guard against a future one rather than a fix for a
# live leak; it costs nothing, because a message with no credential is
# propagated untouched.
return _open_libbigwig_file(
file_url,
lambda: pyBigWig.open(file_url)) # pylint: disable=I1101
@dataclass(frozen=True)
class _StoredFileStat:
"""What one ``info()`` of a stored file is worth to this protocol.
Both fields come out of the same dict, so asking for them separately
is two round trips against one key -- on a remote store, two HEADs.
Bundled so the callers that want both can say so in one call
(gain#936).
The modification time is deliberately NOT here, because no key of
that dict carries it portably. The local filesystem spells it
``mtime`` and s3 ``LastModified``; the memory filesystem has neither
and offers only ``created``, which is a ``datetime`` rather than the
float the recorded state holds. Nor is ``created`` a stand-in for
it: on the local filesystem that is ``st_ctime``, which a chmod
moves and a write does not have to agree with -- measured 0.05s
apart from ``modified()`` on a file that had only been chmod'ed.
Reading it from the dict would therefore mean encoding which key
each backend happens to use, and being wrong about one of them does
not fail loudly: the recorded time would simply stop matching what
:meth:`FsspecRepositoryProtocol.get_resource_file_timestamp` reports,
so every cache verdict would read the file as changed and download
it again. It stays that method's own call, which is the one thing
that reports the value faithfully on every backend.
:meth:`FsspecRepositoryProtocol._get_filepath_timestamp` does fall
back to the dict's ``created``, which reads as a contradiction of
the above and is not one: that branch is reached only where
``modified()`` raises ``NotImplementedError``, i.e. where there is
no faithful answer to prefer to it. Where ``modified()`` answers --
every backend this protocol is used against -- it wins, and this
class is why.
"""
size: int
change_token: str | None
[docs]
class FsspecReadWriteProtocol(
FsspecReadOnlyProtocol, ReadWriteRepositoryProtocol):
"""Provides fsspec genomic resources repository protocol."""
def __init__(
self, proto_id: str,
url: str, *,
filesystem: fsspec.AbstractFileSystem,
public_url: str | None = None,
**kwargs: Any,
):
super().__init__(
proto_id, url,
filesystem=filesystem,
public_url=public_url,
**kwargs,
)
self.filesystem.makedirs(self.url, exist_ok=True)
def _get_resource_file_lockfile_path(
self, resource: GenomicResource, filename: str,
) -> str:
"""Return path of the resource file's lockfile.
Another join of its own, so it repeats the containment check --
``filelock`` creates and truncates the lockfile on acquire
(gain#467).
"""
if self.scheme != "file":
raise NotImplementedError(self._non_local_lock_message())
validate_resource_file_name(resource.resource_id, filename)
resource_url = self.get_resource_url(resource)
path = os.path.join(
resource_url, GRR_INTERNAL_DIR, f"{filename}.lockfile")
return path.removeprefix(f"{self.scheme}://")
def _non_local_lock_message(self) -> str:
return (
f"resource file locking is only supported on a local "
f"filesystem; {self.get_url()} uses the unsupported scheme "
f"<{self.scheme}>")
[docs]
def obtain_resource_file_lock(
self, resource: GenomicResource, filename: str,
timeout: float = -1,
) -> AbstractContextManager:
"""Lock a resource's file.
The lock is a lockfile, which only provides mutual exclusion on a
local filesystem. Off ``file`` this used to return a no-op context
manager -- every caller "acquired" it instantly, so the caching
protocol serialised nothing and concurrent readers saw partially
written files. Refuse rather than hand out a lock that does not
lock; a GRR cache must be local. See #473.
"""
if self.scheme != "file":
raise NotImplementedError(self._non_local_lock_message())
lockfile = self._get_resource_file_lockfile_path(resource, filename)
return FileLock(lockfile, timeout=timeout)
def _scan_path_for_resources(
self, path_array: list[str],
) -> Generator[Any, None, None]:
url = os.path.join(self.url, *path_array)
path = os.path.join(self.root_path, *path_array)
assert isinstance(url, str)
if not self.filesystem.isdir(url):
return
content = []
for direntry in self.filesystem.ls(url, detail=False):
if self.netloc and direntry.startswith(self.netloc):
direntry = direntry[len(self.netloc):]
name = os.path.relpath(direntry, path)
if name.startswith("."):
continue
content.append(name)
if GR_CONF_FILE_NAME in content:
res_path = "/".join(path_array)
# The whole path is parsed here, at the resource, not segment
# by segment on the way down -- so a malformed folder is walked
# and each resource under it is reported and skipped by its
# own full path, rather than the folder once.
try:
resource_id, version = parse_gr_id_version_token(res_path)
except ValueError:
logger.warning(
"repo %s: skipping directory <%s> -- its path %s",
self.proto_id, escape_unsafe_characters(res_path),
malformed_resource_id_reason(res_path))
return
yield resource_id, version, res_path
else:
for name in content:
yield from self._scan_path_for_resources([*path_array, name])
def _scan_resource_for_files(
self, resource_path: str, path_array: list[str],
ancestor_specs: list[tuple[int, pathspec.PathSpec]] | None = None,
) -> Generator[tuple[str, str], None, None]:
"""Yield ``(name, url)`` for every file of a resource.
Whether DVC manages a file is no part of the scan's answer. What a
file IS -- resource data, or a page GAIn generated -- is read off its
path by :meth:`collect_resource_entries`, never off the presence of a
sidecar (#373).
"""
url = os.path.join(self.url, resource_path, *path_array)
if not self.filesystem.isdir(url):
if path_array:
yield os.path.join(*path_array), url
return
path = os.path.join(self.root_path, resource_path, *path_array)
if ancestor_specs is None:
ancestor_specs = []
# Path of the current directory from the GRR root. Each accumulated
# spec is anchored by the depth of the directory that holds its
# .gitignore, so _is_gitignored can match a file against a path
# relative to that .gitignore's own directory (git semantics).
full_parts = [
part for part in resource_path.split("/") if part
] + path_array
# Read the .gitignore in the current directory and add its spec,
# anchored at this directory's depth.
current_specs = ancestor_specs
spec = self._load_gitignore_spec(os.path.join(url, ".gitignore"))
if spec is not None:
current_specs = [*ancestor_specs, (len(full_parts), spec)]
raw_names = []
for direntry in self.filesystem.ls(url, detail=False):
if self.netloc and direntry.startswith(self.netloc):
direntry = direntry[len(self.netloc):]
name = os.path.relpath(direntry, path)
if name.startswith("."):
continue
raw_names.append(name)
# `dvc add <file>` writes `/<file>` into .gitignore and drops a
# sibling `<file>.dvc` pointer; the real data file must stay in the
# manifest. Exempt a gitignored leaf ONLY when it has such a genuine
# pointer in this directory (see _is_dvc_managed_leaf for the exact,
# cli-consistent, crash-safe test). This is done per-candidate and
# lazily: only the sibling `.dvc` of an actually-gitignored leaf is
# ever opened, so a directory with nothing gitignored -- the common
# case -- opens zero `.dvc` files (gain#209).
sibling_names = set(raw_names)
for name in raw_names:
if self._is_gitignored(name, full_parts, current_specs):
# A gitignored leaf named by a genuine sibling `<name>.dvc`
# pointer is a `dvc add <file>` data file and must stay in
# the manifest -- regardless of WHICH gitignore rule ignored
# it. Being ignored by a coincidental ancestor pattern (e.g.
# root `*.tmp` matching `sub/x.tmp`) instead of the `/x.tmp`
# line dvc itself writes is still exactly the DVC situation:
# the pointer proves the data is DVC-managed, so keeping it
# is correct (gain#209).
if not self._is_dvc_managed_leaf(url, name, sibling_names):
continue
yield from self._scan_resource_for_files(
resource_path, [*path_array, name], current_specs)
continue
# A file that is not gitignored is already in the scan, and no
# `.dvc` sibling is consulted for it -- `_is_dvc_managed_leaf`
# opens the sidecar, and a directory with nothing gitignored must
# open none (gain#209).
yield from self._scan_resource_for_files(
resource_path, [*path_array, name], current_specs)
def _is_dvc_managed_leaf(
self, url: str, name: str, sibling_names: set[str],
) -> bool:
"""Return True if gitignored ``name`` is a per-file `dvc add` output.
``name`` is a genuine DVC-managed data file -- and so must stay in the
manifest despite being gitignored -- iff ALL hold (gain#209):
1. ``name`` is not a directory. Only per-file ``dvc add <file>`` is
supported: GAIn cannot verify the ``.dir`` md5 sum of a
``dvc add <dir>`` output against anything it can read, so it
refuses such a resource outright rather than describe it -- see
``repository.collect_dvc_entries``, which fails the command on
the sidecar (gain#255). The directory itself stays out of the
scan.
2. a sibling ``<name>.dvc`` exists in this directory and is a regular
file (a *directory* literally named ``<name>.dvc`` is not a
pointer and must not be opened).
3. that ``<name>.dvc`` parses as a well-formed DVC pointer -- a dict
with an ``outs`` list of dicts -- that declares ``name`` as one of
its outputs (``out["path"] == name``). Both this test and
``repository.collect_dvc_entries`` delegate to
:func:`dvc.parse_dvc_pointer_out`, so the scanner and the
manifest builder cannot classify the same sidecar differently.
Parsing NEVER raises: a binary/non-UTF-8 ``.dvc`` (read in binary and
handed to ``yaml.safe_load`` as bytes, so no UnicodeDecodeError), a
directory opened as a file, or any malformed YAML/shape is treated as
"not a pointer" -- the scan must never abort on stray content.
"""
# (1) per-file dvc only: a gitignored directory is never exempted.
if self.filesystem.isdir(os.path.join(url, name)):
return False
# (2) a sibling pointer must exist and be a regular file.
dvc_name = f"{name}{DVC_SUFFIX}"
if dvc_name not in sibling_names:
return False
dvc_url = os.path.join(url, dvc_name)
if self.filesystem.isdir(dvc_url):
return False
# (3) it must parse as a genuine pointer declaring `name` as output.
try:
with self.filesystem.open(dvc_url, "rb") as infile:
content = cast(bytes, infile.read())
except (OSError, ValueError) as error:
logger.debug(
"ignoring unreadable .dvc pointer %s: %s", dvc_url, error)
return False
return parse_dvc_pointer_out(content, name) is not None
def _load_gitignore_spec(
self, gitignore_url: str,
) -> pathspec.PathSpec | None:
"""Return the PathSpec for a .gitignore, or None if absent/empty."""
if not self.filesystem.exists(gitignore_url):
return None
with self.filesystem.open(gitignore_url, "rt") as f:
raw = cast(str, f.read())
lines = [
line for line in raw.splitlines()
if line and not line.startswith("#")
]
if not lines:
return None
return pathspec.PathSpec.from_lines("gitignore", lines)
def _collect_ancestor_specs(
self, resource_path: str,
) -> list[tuple[int, pathspec.PathSpec]]:
"""Seed gitignore specs from every directory above the resource.
Walk the directories between the GRR root (``self.url``, inclusive)
and the resource directory (exclusive), reading each ``.gitignore``.
The walk never climbs above ``self.url``, so a ``.gitignore`` outside
the GRR is never read. Each spec is anchored by the depth (from the
GRR root) of the directory that holds it, so it matches files under
the resource relative to its ``.gitignore`` root -- as git applies a
repository-root rule to a nested path. The resource's own
``.gitignore`` is read by the descending scan, not here.
Ancestor and descendant specs are OR-combined by :meth:`_is_gitignored`
(any match ignores); cross-level ``!`` negation -- a deeper
``.gitignore`` re-including a file an ancestor ignored -- is NOT
honored. See that method's docstring for the limitation.
"""
parts = [part for part in resource_path.split("/") if part]
specs: list[tuple[int, pathspec.PathSpec]] = []
for depth in range(len(parts)):
spec = self._load_gitignore_spec(
os.path.join(self.url, *parts[:depth], ".gitignore"))
if spec is not None:
specs.append((depth, spec))
return specs
@staticmethod
def _is_gitignored(
name: str,
full_parts: list[str],
specs: list[tuple[int, pathspec.PathSpec]],
) -> bool:
"""Return True if name is excluded by any accumulated gitignore spec.
Specs are combined by OR: ``name`` is dropped if *any* level's
``.gitignore`` matches it. This matches git for the common case of
non-negated patterns, but it does NOT implement git's full
last-match-wins-across-levels semantics: a file ignored by an ancestor
``.gitignore`` cannot be re-included by a ``!pattern`` in a deeper
``.gitignore`` (cross-level negation). Negation within a single
``.gitignore`` still works, since ``pathspec`` resolves last-match
inside one spec. This limitation is characterized by
``test_gitignore_ancestor_negation_across_levels_is_not_honored``.
"""
for anchor_depth, spec in specs:
# Path of `name` relative to the directory that holds this
# .gitignore: drop the leading `anchor_depth` components (GRR root
# -> that directory) from the current path, then add the file name.
rel_parts = [*full_parts[anchor_depth:], name]
rel_path = "/".join(rel_parts)
# Check as both a file and a directory path so that
# trailing-/ patterns (e.g. logs/) prune whole directories.
if spec.match_file(rel_path) or spec.match_file(rel_path + "/"):
return True
return False
def _stat_filepath(self, filepath: str) -> _StoredFileStat:
"""Stat a path once, and read everything that dict is worth.
The single place the ``info()`` dict is interpreted, so the size
a caller gets and the token beside it describe the same answer
from the store rather than two answers taken a round trip apart.
One implementation covers every fsspec scheme: the token is the
``ETag`` out of ``info()``, and a filesystem that reports none --
the local filesystem reports none -- yields None. It is returned
verbatim, quotes and multipart suffix included, because nothing
may depend on its shape.
``info()`` is answered cache-first, so a token is only as fresh
as the listing behind it.
"""
info = self.filesystem.info(filepath)
etag = info.get("ETag")
# Empty as well as absent: s3fs fills a missing ETag on the
# head_object path with "" rather than None, and an empty token
# would compare equal to itself for ever, which is the one way
# this could stop noticing a change altogether.
return _StoredFileStat(
size=int(info["size"]),
change_token=str(etag) if etag else None)
def _get_filepath_change_token(self, filepath: str) -> str | None:
"""The store's own change token for a path, or None if it has none."""
return self._stat_filepath(filepath).change_token
def _get_filepath_timestamp(self, filepath: str) -> float:
try:
modification = self.filesystem.modified(filepath)
modification = modification.replace(tzinfo=datetime.UTC)
return cast(float, round(modification.timestamp(), 2))
except NotImplementedError:
info = self.filesystem.info(filepath)
modification = cast(float, info.get("created"))
return cast(float, round(modification, 2))
[docs]
def collect_all_resources(self) -> Generator[GenomicResource, None, None]:
"""Return generator over all resources managed by this protocol."""
for res_id, res_ver, res_path in self._scan_path_for_resources([]):
res_fullpath = os.path.join(self.root_path, res_path)
assert res_fullpath.startswith("/")
res_fullpath = f"{self.scheme}://{self.netloc}{res_fullpath}"
with self.filesystem.open(
os.path.join(
res_fullpath, GR_CONF_FILE_NAME), "rt") as infile:
config = yaml.safe_load(infile)
manifest: Manifest | None = None
manifest_filename = os.path.join(
res_fullpath, GR_MANIFEST_FILE_NAME)
if self.filesystem.exists(manifest_filename):
with self.filesystem.open(manifest_filename, "rt") as infile:
logger.debug("loading manifest from %s", manifest_filename)
manifest = Manifest.from_file_content(
cast(str, infile.read()))
yield self.build_genomic_resource(
res_id, res_ver, config, manifest)
[docs]
def scan_resource_entries(self, resource: GenomicResource) -> ResourceScan:
"""Scan the resource and return what was found."""
resource_path = resource.get_full_id()
result = Manifest()
unreadable: dict[str, str] = {}
ancestor_specs = self._collect_ancestor_specs(resource_path)
for name, path in self._scan_resource_for_files(
resource_path, [], ancestor_specs):
# The two pages `resource-info` writes are build artefacts, not
# resource data, and stay out of the manifest -- whether or not
# DVC manages them, since both are regenerated on every run.
# Every OTHER file is manifested, whatever its extension: the
# old "drop every name ending in html" rule was a proxy for "this
# is a page GAIn generated" and silently dropped any html file a
# resource legitimately carries as data (#373).
if is_generated_info_page(name):
continue
try:
size = self._get_filepath_size(path)
except OSError as error:
# The listing yielded this name and the stat could not
# describe it. Reported rather than raised: a `.dvc` sidecar
# may still describe it, and only the caller -- which has
# the sidecars -- can tell a garbage-collected DVC cache
# link from a genuinely broken resource (gain#503).
#
# `OSError`, not `FileNotFoundError`: a symlink into a
# shared DVC cache fails to resolve for more reasons than a
# collected cache object -- a loop (ELOOP), a target whose
# parent is not a directory (ENOTDIR), a cache directory
# this run may not traverse (EACCES). They are one situation
# to the user, whose `exists()` is False for every one of
# them, and were one crash each.
#
# Caught rather than pre-tested so that a repository with
# nothing broken pays NOTHING: the happy path is the same
# single stat it always was, and the link probe below runs
# only for a name that already failed.
if self.scheme != "file":
# Only a local filesystem has symlinks. A remote store
# that lists a key and then fails to describe it is far
# likelier to be a transient fault than a steady state,
# and letting a sidecar answer for it would publish an
# md5 sum for an object that is not in the bucket.
raise
reason = self._unreadable_detail(path, error)
logger.debug(
"cannot read <%s> of <%s>: %s",
name, resource.resource_id, reason)
unreadable[name] = reason
continue
result.add(ManifestEntry(name, size, None))
return ResourceScan(result, unreadable)
def _unreadable_detail(self, path: str, error: OSError) -> str:
"""Return why ``path`` could not be read, for a human to read.
For the case this was written for -- a link into a DVC cache that
is no longer resolvable -- naming the link target IS the diagnosis,
and the errno distinguishes a collected cache object from a cache
that is merely unreachable. Only ever called for a name that
already failed to stat, so this never touches the happy path.
"""
reason = error.strerror or type(error).__name__
local_path = path.removeprefix(f"{self.scheme}://")
target: str | None = None
with suppress(OSError):
if os.path.islink(local_path):
target = os.readlink(local_path)
if target is None:
return reason
return f"a symlink to <{target}>: {reason}"
def _enumerate_resources(self) -> Iterable[GenomicResource]:
"""Enumerate by scanning the repository, not by reading ``.CONTENTS``.
The whole of this class's contribution to ``get_all_resources_dict``;
the memo, its lock and the ordering are inherited (#515).
Scanning is not a preference here, it is the only thing that can
work: this class *writes* ``.CONTENTS`` -- ``build_content_file``
enumerates through ``get_all_resources`` and serializes the result --
so reading it back to answer that same enumeration would be circular,
and on a repository that has never been repaired there is no
``.CONTENTS`` to read at all. That is the ``grr_manage repo-repair``
path over a fresh directory.
"""
return self.collect_all_resources()
def _get_resource_file_state_path(
self, resource: GenomicResource, filename: str) -> str:
"""Return filename of the resource file state path.
This joins the resource url itself and so does NOT go through
``get_resource_file_url``; the containment check is repeated here on
purpose -- see gain#467.
"""
validate_resource_file_name(resource.resource_id, filename)
resource_url = self.get_resource_url(resource)
return os.path.join(
resource_url, GRR_INTERNAL_DIR, f"{filename}.state")
def _get_resource_file_download_path(
self, resource: GenomicResource, filename: str) -> str:
"""Return a unique path to download a resource file into.
Inside the resource's ``.grr`` directory, next to the file's
``.state`` and ``.lockfile``: protocol-internal, on the same
filesystem as the resource file itself, and skipped by everything
that enumerates resource files. The ``uuid`` component keeps two
concurrent attempts on the same file from writing the same temp
path. See gain#273.
Another join of its own, so it repeats the containment check --
see gain#467.
"""
validate_resource_file_name(resource.resource_id, filename)
resource_url = self.get_resource_url(resource)
return os.path.join(
resource_url, GRR_INTERNAL_DIR,
f"{filename}.{uuid.uuid4().hex}.part")
[docs]
def publish_raw_file(
self, resource: GenomicResource, filename: str,
mode: str = "wt") -> AbstractContextManager[IO]:
# No directory creation here on purpose: staging the temp already
# makes the whole chain, and doing it here would run at CALL time,
# before the returned context manager is entered -- a side effect
# on a seam whose whole point is that nothing happens unless the
# write completes.
return self._publish_file(
self.get_resource_file_url(resource, filename), mode)
[docs]
def publish_repository_file(
self, filename: str,
mode: str = "wb") -> AbstractContextManager[IO]:
"""Publish one of the repository's own artifacts, by file name.
The counterpart of :meth:`publish_raw_file` for the artifacts that
belong to no resource. The artifacts this protocol builds itself
reach :meth:`_publish_file` directly; this is the way in for one
that is built somewhere else and handed over as bytes -- the FTS
search index, which the CLI assembles from every resource's
implementation and cannot build from in here without importing
the layer above it (gain#948).
Same seam, so the same guarantee: the artifact already published
is replaced by a completed one in a single move, or not at all.
Binary by default, unlike :meth:`publish_raw_file`: a repository
artifact is a built object -- gzipped bytes, a rendered page --
and not the text a resource file usually is.
``filename`` names an artifact of the repository itself and is
joined to its url unvalidated, which is safe only while callers
pass a constant. There is no containment rule to check it against:
the resource-file joins each re-run
:func:`validate_resource_file_name` (gain#467), but that rule is
about staying inside a *resource*, and these artifacts sit beside
the resources rather than in one. A caller that ever wants to
publish a name it did not author needs such a rule written first.
"""
return self._publish_file(
os.path.join(self.url, filename), mode)
def _verify_published_stat(
self, filepath: str,
*, wrote: int, what: str, verb: str,
) -> _StoredFileStat:
"""Check what a move actually landed, and report it if it differs.
The guard gain#880 put on the download's move, shared with the
publish seam gain#933 put the repository's own artifacts behind:
both stat the object the move produced and refuse a size that is
not the one they handed it. Written once so the two cannot drift
-- the discard step beside it is shared for the same reason.
What the two callers keep of their own is only what they can say
truthfully: the download has *verified* its bytes against a
manifest md5 before the move, and a publish has merely *written*
them, so the ``verb`` and the ``what`` prefix are theirs.
Returns the whole stat rather than the size it checked, because
the download path records the change token beside the size and
both come out of the one ``info()`` this makes anyway (gain#936).
A publish ignores the return value entirely, as it always has,
and pays no more for the stat than when this returned an int --
the one ``info()`` behind it is the same call either way.
Takes the path rather than a callable that produces the stat.
It used to take the callable because the two callers reached the
object differently -- a publish through the plain path stat, the
download through :meth:`get_resource_file_size`. Since gain#936
both go through :meth:`_stat_filepath`, so the thunk carried no
variation left and only hid that the two calls are the same one.
"""
try:
published = self._stat_filepath(filepath)
except FileNotFoundError as err:
raise CorruptedPublishError(
f"{what}; {verb} {wrote} bytes, "
f"published nothing") from err
if published.size != wrote:
raise CorruptedPublishError(
f"{what}; {verb} {wrote} bytes, "
f"published {published.size}")
return published
def _get_file_publish_path(self, filepath: str) -> str:
"""Return a unique path to publish ``filepath`` through.
Inside a ``.grr`` directory beside the target, so the temp is
always on the same filesystem as the file it will become and the
move stays a rename wherever the scheme has one. For a resource
file that is the resource's own ``.grr``, next to the ``.state``
and ``.lockfile`` a download already stages in; for one of the
repository's own artifacts -- the contents index, the index page,
the rendered ``about.html``, which belong to no resource -- it is
the repository root's. One rule covers both.
Nothing enumerates what lands there: a name starting with "." is
passed over by both the resource walk and the file scan, so a
temp in flight is invisible to a repository scan and to a manifest
build alike.
The ``uuid`` component keeps two concurrent publishes of the same
artifact from writing the same temp path, for the same reason
:meth:`_get_resource_file_download_path` carries one.
Deliberately NOT that method's rule, though, and the difference
shows on a nested name: a download of ``sub/data.txt`` stages
under the resource's single ``.grr``, while publishing it stages
at ``<resource>/sub/.grr/`` -- always beside the target, which is
what keeps the temp on the target's own filesystem whether the
target is a resource file or one of the repository's own
artifacts. Both are contained and both are skipped by the walks;
only the location differs.
The directory outlives the temp: it is created on demand and not
removed, so a repository accumulates an empty ``.grr`` at its root
and at each directory published into. Empty directories are
nothing to git, which is what the published GRRs are, but a
repository copied by other means carries them along. See gain#933.
"""
return os.path.join(
os.path.dirname(filepath), GRR_INTERNAL_DIR,
f"{os.path.basename(filepath)}.{uuid.uuid4().hex}.part")
@contextmanager
def _publish_file(
self, filepath: str, mode: str,
*, encoding: str | None = None,
) -> Generator[IO, None, None]:
"""Yield a handle whose bytes reach ``filepath`` only if it closes.
The repository's own artifacts used to be opened at their live
path and written in place, so an interrupted write left a
truncated artifact where every consumer reads one and nothing to
roll back to -- a truncated ``.CONTENTS.json.gz`` breaks the whole
repository for every client, and ``.MANIFEST`` is what every
download-path md5 check compares against (gain#933).
This is gain#273's write-temp-verify-move, hoisted off the
download path so the repository's own writes get it too: the
caller writes into a private temp, and the temp is moved onto the
target only once the handle has closed cleanly. A failure
anywhere before the move -- an open that never lands, a write that
fails mid-stream, a close that does, an interrupt -- leaves the
previously published artifact exactly as it was, and leaves no
temp behind either. On a local filesystem the move is an atomic
rename; on an object store it degrades to a copy-and-delete,
which loses the atomicity but keeps the verify-then-publish
order -- one code path for every scheme, as on the download side.
Because the check lives on the move, gain#880's post-move
verification comes with it: the published object is stated and
must match what was written, so a move that lands nothing or
lands a short object is reported rather than published silently.
Only size-changing corruption is caught, for the reason
:class:`CorruptedPublishError` gives.
What that check buys here is a report, not a repair. Past the
move the previous artifact is gone and there is nothing to roll
back to, and unlike ``copy_resource_file`` -- which retries a
corrupting publish until the object lands intact -- the callers
of this seam publish once and propagate. A move-time corruption
therefore leaves the repository needing a republish; it is only
guaranteed not to leave it needing one *silently*.
One further difference from writing in place, harmless for the
artifacts published today but real: the move replaces the inode,
so the published file takes a fresh identity rather than
inheriting the mode of the file it replaces, and a target that
was a symlink is replaced by a regular file instead of being
written through. The download path has always had this property.
"""
# A mode that is not a write would be actively destructive here:
# the handle is opened on an EMPTY temp, so "at" appends to
# nothing and the move then replaces the artifact with just the
# appended part. Refused rather than documented, because the
# damage is silent and this is a public seam that will acquire
# callers.
if "w" not in mode:
raise ValueError(
f"publishing {filepath} needs a write mode, got {mode!r}: "
f"a publish replaces the file, it cannot extend it")
tmp_filepath = self._get_file_publish_path(filepath)
# Unconditional: ``exist_ok`` already tolerates the directory being
# there, so a preceding ``exists`` only adds a round trip to the
# common case. That matters because ``save_manifest`` publishes
# once per resource, so this is paid per resource across a whole
# repository -- on an object store, per resource over the network.
self.filesystem.makedirs(os.path.dirname(tmp_filepath), exist_ok=True)
open_kwargs: dict[str, Any] = {}
if encoding is not None:
open_kwargs["encoding"] = encoding
moved = False
try:
# Redacted like every other open on this protocol: the sinks
# that reach here used to go through ``_open_fsspec_file``,
# and a publish to a credential-bearing url must not put the
# userinfo into a traceback just because the write now stages.
#
# The handle is wrapped, not merely opened redacted. The sink
# yielded below is written to by the caller, and a store finishes
# a write on release -- so redacting the open alone would leave
# this the one write path in the protocol still able to surface
# the fetch url, which is exactly the shape gain#1078 closed
# everywhere else. See ADR 0023.
handle = cast("IO", _RedactingFile(_run_redacting_url_credentials(
lambda: self.filesystem.open(
tmp_filepath, mode, **open_kwargs))))
with handle as outfile:
yield outfile
written = self._get_filepath_size(tmp_filepath)
self.filesystem.mv(tmp_filepath, filepath)
moved = True
self._verify_published_stat(
filepath,
wrote=written,
what=f"artifact publish is corrupt {filepath}",
verb="wrote")
finally:
if not moved:
self._discard_unpublished_file(tmp_filepath)
[docs]
def get_resource_file_timestamp(
self, resource: GenomicResource, filename: str) -> float:
url = self.get_resource_file_url(resource, filename)
return self._get_filepath_timestamp(url)
[docs]
def get_resource_file_change_token(
self, resource: GenomicResource, filename: str) -> str | None:
url = self.get_resource_file_url(resource, filename)
return self._get_filepath_change_token(url)
def _get_filepath_size(
self, filepath: str) -> int:
return self._stat_filepath(filepath).size
[docs]
def get_resource_file_size(
self, resource: GenomicResource, filename: str) -> int:
path = self.get_resource_file_url(resource, filename)
return self._get_filepath_size(path)
[docs]
def save_resource_file_state(
self, resource: GenomicResource, state: ResourceFileState) -> None:
"""Save resource file state into internal GRR state."""
path = self._get_resource_file_state_path(resource, state.filename)
# Unconditional -- see ``_publish_file`` and gain#1042.
self.filesystem.makedirs(os.path.dirname(path), exist_ok=True)
content = asdict(state)
with self.filesystem.open(path, "wt", encoding="utf8") as outfile:
outfile.write(yaml.safe_dump(content))
[docs]
def load_resource_file_state(
self, resource: GenomicResource,
filename: str) -> ResourceFileState | None:
"""Load resource file state from internal GRR state.
If the specified resource file has no internal state returns None.
"""
path = self._get_resource_file_state_path(resource, filename)
if not self.filesystem.exists(path):
return None
with self.filesystem.open(path, "rt", encodings="utf8") as infile:
content = yaml.safe_load(infile.read())
if content is None or not content:
return None
return ResourceFileState(
content["filename"],
content["size"],
content["timestamp"],
content["md5"],
# Written by every state since gain#881, and by none
# before it: a state file already on disk carries no
# token and must keep loading, falling back to the
# modification time until it is next rebuilt.
content.get("change_token"),
)
[docs]
def delete_resource_file(
self, resource: GenomicResource, filename: str) -> None:
"""Delete a resource file and it's internal state."""
filepath = self.get_resource_file_url(resource, filename)
if self.filesystem.exists(filepath):
self.filesystem.delete(filepath)
statepath = self._get_resource_file_state_path(resource, filename)
if self.filesystem.exists(statepath):
self.filesystem.delete(statepath)
[docs]
def copy_resource_file(
self,
remote_resource: GenomicResource,
dest_resource: GenomicResource,
filename: str,
on_bytes: Callable[[int], None] | None = None,
) -> ResourceFileState | None:
"""Copy a resource file into repository.
A transient stall or drop mid-download (common when fetching a large
resource over a slow HTTP GRR link) is retried from scratch with
exponential backoff rather than aborting the file. See gain#43.
``on_bytes``, when given, is called with the number of bytes written
for each chunk during the download (see gain#77). Because a retried
attempt re-downloads the whole file from scratch, the bytes credited
by a failed attempt are rolled back with a single compensating
negative call before the retry, so a caller-side byte counter never
double-counts.
"""
assert dest_resource.resource_id == remote_resource.resource_id
logger.debug(
"copying resource file (%s: %s) from %s",
remote_resource.resource_id, filename,
remote_resource.proto.proto_id)
remote_manifest = remote_resource.get_manifest()
if filename not in remote_manifest:
self.delete_resource_file(dest_resource, filename)
return None
manifest_entry = remote_manifest[filename]
dest_filepath = self.get_resource_file_url(dest_resource, filename)
dest_parent = os.path.dirname(dest_filepath)
# Unconditional -- see ``_publish_file`` and gain#1042.
#
# ``makedirs``, not the ``mkdir`` this site used to call:
# ``AbstractFileSystem.mkdir`` takes no ``exist_ok``, and the local
# and memory backends raise ``FileExistsError`` from their own
# is-it-there check -- ``exists()`` for the one, a lookup in its
# store for the other -- before any keyword is consulted. The
# ``exist_ok=True`` passed here was therefore inert, which made the
# guard in front of it load-bearing at this site alone. s3 tolerates
# either spelling, so it cannot show the difference.
self.filesystem.makedirs(dest_parent, exist_ok=True)
# Bytes credited to on_bytes during the current attempt, so a
# retryable failure can roll them back before the next attempt.
attempt_bytes = 0
def tracking_on_bytes(n: int) -> None:
nonlocal attempt_bytes
attempt_bytes += n
assert on_bytes is not None
on_bytes(n)
wrapped_on_bytes = (
tracking_on_bytes if on_bytes is not None else None)
last_error: BaseException | None = None
for attempt in range(1, _COPY_MAX_ATTEMPTS + 1):
attempt_bytes = 0
try:
return self._download_resource_file(
remote_resource, dest_resource, filename,
dest_filepath, manifest_entry.md5,
expected_size=manifest_entry.size,
on_bytes=wrapped_on_bytes)
except _RETRYABLE_COPY_ERRORS as error:
last_error = error
if on_bytes is not None and attempt_bytes:
# roll back the partially-credited bytes of this attempt
on_bytes(-attempt_bytes)
if attempt >= _COPY_MAX_ATTEMPTS:
break
delay = _COPY_BACKOFF_BASE * (3 ** (attempt - 1))
# A read failure arrives already redacted by the handle; a
# failure of this clause's own machinery -- the temp-file
# open, the publish -- arrives as the store rendered it.
logger.warning(
"transient failure downloading (%s: %s): %s; "
"retrying in %ss (attempt %s/%s)",
dest_resource.resource_id, filename,
strip_url_credentials(str(error)),
delay, attempt + 1, _COPY_MAX_ATTEMPTS)
time.sleep(delay)
assert last_error is not None
# A last redaction on the way out, for a ``last_error`` that reached
# here without passing through the handle -- one raised by the
# ``except`` clause's own machinery rather than by a read.
#
# This used to be the ONLY place the download path could redact, and
# the rule was positional: never before the ``except`` above, because
# that clause classifies by exception type and a rebuilt error is a
# different type. ``ClientResponseError`` cannot be reconstructed
# from a message alone, so it rebuilt to a bare ``OSError``, which
# matches nothing in ``_RETRYABLE_COPY_ERRORS`` -- redacting earlier
# would have cut the retry budget to a single attempt for exactly the
# authed downloads this protects (gain#620).
#
# gain#1078 made the reads under this loop redact too, so "earlier"
# now happens on every attempt. The rule it relied on has been made
# structural instead: ``_rebuild_error_without_url_credentials``
# preserves retryability, rebuilding a transient failure it cannot
# reconstruct as ``RetryableCopyError`` rather than ``OSError``. The
# retry budget survives redaction wherever redaction happens.
raise _error_without_url_credentials(last_error)
def _download_resource_file(
self,
remote_resource: GenomicResource,
dest_resource: GenomicResource,
filename: str,
dest_filepath: str,
expected_md5: str | None,
*,
expected_size: int,
on_bytes: Callable[[int], None] | None = None,
) -> ResourceFileState:
"""Download a single resource file once, verify it, then publish it.
Opens a fresh remote handle and streams into a private temp file in
the resource's ``.grr`` directory; the file is moved to its real
path only once it has been verified, so the repository never holds
an unverified resource file and an attempt that fails *before* the
move leaves nothing behind at the real path (gain#273). An attempt
that fails *after* the move does leave the bad object there --
there is nowhere else for it to go by then; what the publish check
buys is that such an object is never recorded as good, so the
retry republishes over it and, failing that, the next cache
verdict fetches it again. On a local filesystem the move is an
atomic rename; on an object store it degrades to a copy-and-delete,
which loses the atomicity but keeps the verify-then-publish order
-- one code path for every scheme.
The download is verified twice: the number of bytes written must
equal the manifest's recorded size (a silent short read in the
fsspec range-reassembly layer ends the stream early and would
otherwise produce a truncated file that only fails at the md5 check
-- gain#292), and the md5 of the written bytes must match the
manifest. The size check runs first so a truncation is reported as
such, with both byte counts, rather than as an opaque checksum
mismatch.
Both of those checks describe the temp file; the move itself is
verified separately, once it has happened -- see
:class:`CorruptedPublishError`.
``on_bytes``, when given, is called with the length of each chunk
right after it is written, to drive a byte-level progress bar
(see gain#77).
"""
tmp_filepath = self._get_resource_file_download_path(
dest_resource, filename)
tmp_parent = os.path.dirname(tmp_filepath)
# Unconditional -- see ``_publish_file`` and gain#1042.
self.filesystem.makedirs(tmp_parent, exist_ok=True)
moved = False
try:
bytes_written = 0
with remote_resource.open_raw_file(
filename, "rb",
uncompress=False) as infile, \
self.filesystem.open(tmp_filepath, "wb") as outfile:
md5_hash = hashlib.md5() # ruff: ignore[hashlib-insecure-hash-function]
while chunk := infile.read(self.CHUNK_SIZE):
outfile.write(chunk)
bytes_written += len(chunk)
if on_bytes is not None:
on_bytes(len(chunk))
md5_hash.update(chunk)
md5 = md5_hash.hexdigest()
if not self.filesystem.exists(tmp_filepath):
# Redacted like the cleanup's own log line: a bare
# ``OSError`` is not retryable, so this escapes the retry
# loop's redacting epilogue entirely, and ``tmp_filepath``
# derives from the credential-bearing fetch url on a write
# protocol over an authed store (gain#620).
raise OSError(
"destination file not created "
f"{strip_url_userinfo(tmp_filepath)}")
if bytes_written != expected_size:
raise TruncatedDownloadError(
f"file copy is truncated "
f"{dest_resource.resource_id} ({filename}); "
f"received {bytes_written} bytes, "
f"expected {expected_size}")
if md5 != expected_md5:
raise ChecksumMismatchError(
f"file copy is broken "
f"{dest_resource.resource_id} ({filename}); "
f"received {bytes_written} bytes (size ok); "
f"md5sum are different: "
f"{md5}!={expected_md5}")
self.filesystem.mv(tmp_filepath, dest_filepath)
moved = True
# The stat is the one the state needs anyway, so checking the
# move costs nothing -- see :class:`CorruptedPublishError`.
published = self._verify_published_stat(
dest_filepath,
wrote=bytes_written,
what=(
f"file publish is corrupt "
f"{dest_resource.resource_id} ({filename})"),
verb="verified")
finally:
if not moved:
self._discard_unpublished_file(tmp_filepath)
# Every field of the state is supplied, so the build reads
# nothing back: the md5 is the one hashed off the bytes as they
# were written, the size and the change token are the two halves
# of the stat that verified the move, and only the modification
# time is a call of its own -- the one field an ``info()`` dict
# cannot be trusted for (see :class:`_StoredFileStat`). That
# leaves the state two metadata round trips per published file
# where it took four, which a full GRR sync pays once per file
# (gain#936).
#
# Two is this state build's budget, not the download's: the
# directory guards above and the move itself still stat, and
# measured against s3 the whole per-file download is nine
# requests rather than two. See gain#1042 for the largest of
# what is left.
state = self.build_resource_file_state(
dest_resource,
filename,
md5=md5,
size=published.size,
change_token=published.change_token,
timestamp=self.get_resource_file_timestamp(
dest_resource, filename))
self.save_resource_file_state(dest_resource, state)
return state
def _discard_unpublished_file(self, tmp_filepath: str) -> None:
"""Remove the temp file of a write that never reached the move.
Called for every way out of :meth:`_download_resource_file` and of
:meth:`_publish_file` that does not reach the move -- a checksum
mismatch, a stalled read, a write that failed mid-stream, an
interrupt -- so no attempt leaves a partial behind. A publish that
the move itself corrupted is past this point: the temp file is
already gone and the bad object sits at the real path, where only
a retry or the next cache verdict can replace it.
The temp file may not exist at all (the remote handle can fail
before the first write), and a removal that fails must not replace
the failure that got us here: the retry loop classifies the error
it sees, and a cleanup error in its place would be neither
retryable nor true.
"""
try:
self.filesystem.rm(tmp_filepath)
except FileNotFoundError:
pass
except Exception as error: # ruff: ignore[blind-except] pylint: disable=broad-except
# Deliberately no ``exc_info``. This runs from the download's
# ``finally``, so the failure that got us here is still
# propagating, and ``exc_info`` renders the whole ACTIVE chain --
# which on an authed GRR is an aiohttp error carrying the
# credential-bearing fetch url. That put the secret in the log by
# a second route, one the retry loop's own redaction never sees
# (gain#620). The cleanup error's own message is what this line
# is about; it and the path are redacted for the same reason,
# the path as the display url it is.
logger.warning(
"unable to remove the unpublished temp file %s: %s",
strip_url_userinfo(tmp_filepath),
strip_url_credentials(str(error)))
[docs]
def classify_resource_file(
self, remote_resource: GenomicResource,
dest_resource: GenomicResource,
filename: str) -> FileCacheVerdict:
"""Decide whether a resource file needs (re)downloading.
This is the lock-free decision half of :meth:`update_resource_file`:
it performs the same checks and the same state-refresh side effect
(rebuild + save the ``.state`` on a missing state or one that no
longer describes the stored file, and delete a file no longer in
the remote manifest), but it never copies/downloads. The
verdict's ``size`` is the manifest byte size for files that will
download (0 otherwise). See gain#78.
The one question it opens with -- is the file there at all -- is
asked as a stat rather than as a boolean, because the same dict
carries the size and the change token a rebuilt state needs. So
the rebuild reads only what that dict cannot say: the
modification time, and the md5 off the bytes themselves. Asking
for a bool and then rebuilding from scratch asked the store
about one key five times where twice will do (gain#1039). Only
the rebuild is cheaper for it: a verdict that finds its recorded
state current spends what it always did, one stat either way.
A stat that fails for a reason other than the file being absent
now reaches the caller instead of reading as "not cached". That
is the one thing ``exists()`` did that this does not: fsspec's
base implementation answers False to *every* exception, so an
unreadable cache directory used to be answered with a download
that was going to fail on the same directory a moment later. On
s3 it is not even a change -- s3fs's own ``exists`` swallows
only ``FileNotFoundError``.
The stat is taken before the md5 that is recorded beside it, so
a file rewritten in between is recorded with the older token
beside the newer digest. That pairing is self-correcting rather
than a lost update: the next verdict reads the token, finds it
moved, and rebuilds. It is the safe half of the ordering -- a
token read *after* the digest would pair a fresh token with a
superseded md5, and nothing afterwards would notice.
"""
assert dest_resource.resource_id == remote_resource.resource_id
remote_manifest = remote_resource.get_manifest()
url = self.get_resource_file_url(dest_resource, filename)
try:
stored = self._stat_filepath(url)
except FileNotFoundError:
size = (
remote_manifest[filename].size
if filename in remote_manifest else 0)
return FileCacheVerdict(needs_download=True, size=size)
local_state = self.load_resource_file_state(dest_resource, filename)
if local_state is None or not self._state_describes_stored_file(
dest_resource, local_state):
local_state = self.build_resource_file_state(
dest_resource, filename,
size=stored.size, change_token=stored.change_token)
self.save_resource_file_state(dest_resource, local_state)
if filename not in remote_manifest:
self.delete_resource_file(dest_resource, filename)
return FileCacheVerdict(needs_download=False, size=0)
manifest_entry = remote_manifest[filename]
if local_state.md5 != manifest_entry.md5:
return FileCacheVerdict(
needs_download=True, size=manifest_entry.size)
return FileCacheVerdict(needs_download=False, size=0)
[docs]
def update_resource_file(
self, remote_resource: GenomicResource,
dest_resource: GenomicResource,
filename: str) -> ResourceFileState | None:
"""Update a resource file into repository if needed."""
verdict = self.classify_resource_file(
remote_resource, dest_resource, filename)
if verdict.needs_download:
return self.copy_resource_file(
remote_resource, dest_resource, filename)
# No download needed: a file deleted because it left the remote
# manifest has no state to return (load returns None); an up-to-date
# file returns its current persisted state.
return self.load_resource_file_state(dest_resource, filename)
def _manifest_for_repository_index(
self, res: GenomicResource,
failed: frozenset[str]) -> Manifest | None:
"""The manifest to publish for ``res`` in repository-wide files.
A resource this run FAILED to verify must not have its manifest
rebuilt from scratch here: that fallback hashes the drifted
bytes, writes a state and publishes an md5 the run had just
refused to record, dropping any pointer-only entry on the way.
Publish the manifest it already had committed, or -- if it
never had one -- leave it out of the repository index entirely
(#373).
"""
if res.resource_id in failed:
try:
return self.load_manifest(res)
except FileNotFoundError:
return None
try:
return res.get_manifest()
except UnsupportedDvcDirectoryOutputError as err:
# The walk met a manifest-less `dvc add <dir>` resource the
# command never selected, and the fallback build refused it
# (#284, via #721's sidecar collection). That is THIS
# resource's failure alone: it is left out of the index, with
# a report, and the walk goes on -- one refused resource must
# not unpublish the healthy ones (the gain#503 shape). A
# committed manifest, had it one, would have been loaded
# above the refusal, so there is nothing older to fall back
# to here.
# The refusal message is complete on its own; a traceback
# would only bury it, hence `error`, not `exception`.
logger.error( # ruff: ignore[error-instead-of-exception]
"not publishing <%s> in the repository index: %s",
res.resource_id, err)
return None
[docs]
def build_content_file(
self, failed: frozenset[str] = frozenset(),
) -> list[dict[str, Any]]:
"""Build the content of the repository (i.e '.CONTENTS.json.gz').
``failed`` names resources this run could not verify; each is
published from the manifest it already had, or left out if it
never had one, so a failed run never rebuilds a manifest from
scratch and poisons the contents with it (#373).
Only the gzipped index is written. An uncompressed
``.CONTENTS.json`` left by an older release is reported rather
than deleted (#758).
"""
content = []
for res in self.get_all_resources():
manifest = self._manifest_for_repository_index(res, failed)
if manifest is None:
continue
content.append({
"full_id": res.get_full_id(),
"id": res.resource_id,
"version": res.get_version_str(),
"config": res.get_config(),
"manifest": manifest.to_manifest_entries(),
})
content = sorted(content, key=operator.itemgetter("id"))
content_filepath = os.path.join(
self.url, GR_CONTENTS_FILE_NAME)
# gzip header OS byte (offset 9) is normalised to 0xff
# ("unknown") so the file is byte-deterministic across
# Python distributions. Upstream CPython hardcodes 0xff,
# but Debian's Python patches gzip.compress to emit 0x03
# ("Unix"), which means the same input produces different
# bytes between a conda Python and a python:3.x-slim
# container — enough to flag .CONTENTS.json.gz as modified
# under `git status --porcelain` in CI even when the JSON
# payload is identical.
gz = gzip.compress(
json.dumps(
content, indent=2, sort_keys=True).encode("utf8"),
mtime=0)
gz = gz[:9] + b"\xff" + gz[10:]
with self._publish_file(content_filepath, "wb") as outfile:
outfile.write(gz)
# Left where it is rather than deleted: in the GRRs that carry
# one it is a tracked file, and a publish has no business
# authoring that deletion in someone else's git tree.
abandoned = os.path.join(self.url, GR_LEGACY_CONTENTS_FILE_NAME)
if self.filesystem.exists(abandoned):
logger.warning(
"%s is stale: the repository index is published gzipped "
"only, so whatever an older release left there has just "
"been left behind by this publish. Only %s is maintained; "
"the stale file can be deleted.",
abandoned, GR_CONTENTS_FILE_NAME)
return content
[docs]
def build_index_info(
self,
repository_template: str = "grr_index.jinja",
about_template: str | None = "grr_about.jinja",
failed: frozenset[str] = frozenset(),
) -> dict:
"""Build info dict for the repository.
``failed`` names resources this run could not verify; each is
described from the manifest it already had, or left off the
index page if it never had one, so the page never triggers a
build-from-scratch of a failed resource's manifest (#373).
"""
result = {}
# In the order the page keeps its rows in -- its own ID sorter's
# -- because until the search index loads, and for good if it
# never does, the published order is all a reader has
# (gain#1351). The memo's own order is left alone.
for res in sorted(
self.get_all_resources(),
key=lambda r: resource_id_collation_key(r.get_full_id())):
manifest = self._manifest_for_repository_index(res, failed)
if manifest is None:
continue
res_size = convert_size(
sum(f for _, f in manifest.get_files()),
)
assert res.config is not None
result[res.get_full_id()] = {
"res_full_id": res.get_full_id(),
"res_id": res.resource_id,
**res.config,
"res_version": res.get_version_str(),
"res_files": len(list(manifest.get_files())),
"res_size": res_size,
"res_summary": res.get_summary(),
}
about_md_path = os.path.join(self.url, "about.md")
has_about = self.filesystem.exists(about_md_path)
about_html_content = ""
if has_about:
with self.filesystem.open(
about_md_path, "rt", encoding="utf8") as infile:
about_md_raw = infile.read()
try:
about_html_content = markdown(about_md_raw)
except Exception as e: # pylint: disable=broad-exception-caught
logger.exception(
"Error occurred while converting about.md to HTML for %s",
self.get_url(),
)
raise ValueError from e
with self._publish_file(
os.path.join(self.url, "about.html"), "wt", encoding="utf8",
) as outfile:
if about_template is not None:
outfile.write(get_template(about_template).render(
about_contents=about_html_content))
else:
outfile.write(about_html_content)
sqlite3_hash = ""
gz_path = os.path.join(self.url, GR_SQLITE_META_FILE_NAME)
if self.filesystem.exists(gz_path):
with self.filesystem.open(gz_path, "rb") as gz_file:
gz_bytes: bytes = cast(bytes, gz_file.read())
sqlite3_hash = hashlib.md5(gz_bytes).hexdigest() # ruff: ignore[hashlib-insecure-hash-function]
# Before the page: the page imports these by relative URL, and a
# page that is published ahead of what it imports has a window
# in which it renders with no search (gain#1335).
self._publish_static_assets()
content_filepath = os.path.join(self.url, GR_INDEX_FILE_NAME)
with self._publish_file(
content_filepath, "wt", encoding="utf8") as outfile:
outfile.write(get_template(repository_template).render(
data=result,
has_about=has_about,
sqlite3_hash=sqlite3_hash,
))
return result
def _publish_static_assets(self) -> None:
"""Publish the files the pages load from the repository.
What gain ships for the pages -- the search engine the index
runs on, and the fonts every page sets its text and icons in --
is published into every repository beside them, so a repository
carries everything its pages need (gain#1335, gain#1400).
Which files is ``gain.templates.static_assets``' business; this
names none. Through :meth:`publish_repository_file`, so each
file lands in a single move or not at all; the names are
constants of gain's own, which is that seam's condition.
Skipped when the published bytes already match. The page beside
them is rewritten every run, but the page changes with the
repository while these change only with gain: rewriting 1.2 MB
of identical bytes would give the mirrors, and the git
repositories the published GRRs live in, a change to notice on
every run.
"""
for filename, content in repository_static_files():
if self._published_bytes(filename) == content:
continue
with self.publish_repository_file(filename) as outfile:
outfile.write(content)
def _published_bytes(self, filename: str) -> bytes | None:
"""The bytes of one of the repository's own artifacts, or None.
One read rather than exists + open + read -- on an object store
each of those is a round trip -- and a redacted one, so a
mid-read failure on an authed url does not surface the
credential (ADR 0023). None for an artifact never published;
the redaction rebuilds an error under its own type, so the
not-found case is still recognisable after it.
"""
filepath = os.path.join(self.url, filename)
try:
return cast(bytes, _run_redacting_url_credentials(
lambda: self.filesystem.cat_file(filepath)))
except FileNotFoundError:
return None
[docs]
def build_local_resource(
dirname: str, config: dict[str, Any]) -> GenomicResource:
"""Build a resource from a local filesystem directory."""
proto = build_fsspec_protocol("d", dirname)
return GenomicResource(".", (0, ), proto, config)
def _basic_auth_header(user: str, password: str) -> str:
"""Return the ``Authorization`` header value for HTTP Basic auth.
The credential is encoded as UTF-8, per RFC 7617, which also forbids
a ``:`` in the user -- raises ``ValueError`` on one.
"""
if ":" in user:
raise ValueError(
'A ":" is not allowed in the HTTP basic auth user (RFC 7617)')
# Encoded here so that no aiohttp auth helper is needed (#1395).
token = base64.b64encode(f"{user}:{password}".encode()).decode("ascii")
return f"Basic {token}"
def _build_filesystem(
url: str, **kwargs: Any,
) -> fsspec.AbstractFileSystem:
# pylint: disable=import-outside-toplevel
# A keyword read here is a keyword that configures a protocol, so adding
# one means adding it to ``_FILESYSTEM_KWARGS`` too -- otherwise a rebuild
# may silently change it under the protocol's holders (#514).
parsed_url = urlparse(url)
if parsed_url.scheme in {"file", ""}:
from fsspec.implementations.local import LocalFileSystem
return LocalFileSystem()
if parsed_url.scheme in {"http", "https"}:
import aiohttp
from fsspec.implementations.http import HTTPFileSystem
base_url = kwargs.get("base_url")
# Relax aiohttp's default 300s total read timeout: a large GRR
# resource (e.g. the ~15GB genome-wide gnomAD file) legitimately
# downloads for far longer. total=None lifts the overall cap while
# sock_read/sock_connect still turn a genuinely stalled read or hung
# connect into a (retryable) error rather than killing the run. See
# gain#43.
client_kwargs: dict[str, Any] = {
"base_url": base_url,
"timeout": aiohttp.ClientTimeout(
total=None, sock_read=120, sock_connect=60),
}
user = kwargs.get("user")
password = kwargs.get("password")
# aiohttp refuses a request that carries credentials both in the url
# userinfo and in an ``Authorization`` header, so a url that already
# embeds them is left to authenticate on its own.
if (user is not None and password is not None
and not _url_carries_userinfo(url)):
client_kwargs["headers"] = {
"Authorization": _basic_auth_header(user, password),
}
return HTTPFileSystem(client_kwargs=client_kwargs)
if parsed_url.scheme == "s3":
from s3fs.core import S3FileSystem
endpoint_url = kwargs.get("endpoint_url")
return S3FileSystem(
anon=False, client_kwargs={"endpoint_url": endpoint_url})
if parsed_url.scheme == "memory":
from fsspec.implementations.memory import MemoryFileSystem
return MemoryFileSystem()
raise NotImplementedError(f"unsupported schema {parsed_url.scheme}")
FsspecRepositoryProtocol = FsspecReadOnlyProtocol | FsspecReadWriteProtocol
_FSSPEC_PROTOCOLS: dict[tuple[str, str], FsspecRepositoryProtocol] = {}
#: The constructions currently in flight, one per key at most. A key is here
#: OR in ``_FSSPEC_PROTOCOLS``, never both: an instance moves across when its
#: whole ``__init__`` chain (or ``__setstate__``) has returned, which is what
#: keeps an unconfigured protocol from ever being reachable (#527).
_FSSPEC_PROTOCOLS_UNDER_CONSTRUCTION: dict[
tuple[str, str], _ProtocolConstruction] = {}
#: Serialises every look at the two dicts above. Held only across the memo
#: read and the bookkeeping that answers it -- never across ``__init__``,
#: which for a read-write protocol does filesystem I/O and would otherwise
#: put every protocol construction in the process behind one remote round
#: trip (#527).
_FSSPEC_PROTOCOLS_GUARD = Lock()
#: The string spellings of a boolean this module accepts for ``read_only``,
#: matching what the definition models coerce and what yaml would have
#: produced unquoted. Kept deliberately closed: a value outside it is a
#: mistake to report, not a value to guess at.
_FALSE_SPELLINGS = frozenset({"false", "no", "off", "0", ""})
_TRUE_SPELLINGS = frozenset({"true", "yes", "on", "1"})
def _resolve_read_only(*, value: str | bool | None) -> bool | None:
"""Return ``read_only`` as a boolean, or ``None`` if it was not passed.
``read_only`` is documented as the one boolean keyword, but it reaches
this module as a *string* from two directions, and a bare truthiness test
read every one of them -- ``"false"`` included -- as read-only, which is
the inversion of what was asked for (#528):
* ``grr_manage --extra-args read_only=false`` parses into ``dict[str,
str]`` and is splatted in verbatim;
* a repository definition is built from the **raw** definition dict, not
from the validated model, so a quoted ``read_only: "false"`` in yaml
arrives here as ``"false"`` even though ``FileRepoDefinition`` coerced
it to ``False`` when the definition was checked.
A value that spells no boolean at all raises rather than defaulting: it
can only be a mistake, and every way of guessing at it silently produces
a repository in a mode nobody asked for.
"""
if value is None or isinstance(value, bool):
return value
spelling = value.strip().lower()
if spelling in _FALSE_SPELLINGS:
return False
if spelling in _TRUE_SPELLINGS:
return True
spellings = sorted(_TRUE_SPELLINGS | (_FALSE_SPELLINGS - {""}))
raise ValueError(
f"read_only must be a boolean, not {value!r}; write one of "
f"{spellings} if it has to be spelled as a string")
[docs]
def build_fsspec_protocol(
proto_id: str, root_url: str, **kwargs: str | bool | None,
) -> FsspecRepositoryProtocol:
"""Create fsspec GRR protocol based on the root url.
``read_only`` is the one boolean among the keyword arguments -- hence the
widened value type; every other keyword is a url or a credential. It is
absent by default rather than ``False`` so that *asking* for a read-write
protocol can be told apart from not asking at all: the two mean different
things on an http(s) url, where only one of them is serviceable (#528).
"""
# pylint: disable=import-outside-toplevel
public_url = cast("str | None", kwargs.pop("public_url", None))
read_only = _resolve_read_only(value=kwargs.pop("read_only", None))
filesystem = _build_filesystem(root_url, **kwargs)
url = urlparse(root_url)
if url.scheme in {"file", "", "s3", "memory"}:
if read_only:
return FsspecReadOnlyProtocol(
proto_id, root_url,
filesystem=filesystem,
public_url=public_url,
**kwargs)
return FsspecReadWriteProtocol(
proto_id, root_url,
filesystem=filesystem,
public_url=public_url,
**kwargs)
if url.scheme in {"http", "https"}:
if read_only is not None and not read_only:
# Asked for read-write over a scheme that cannot serve it. The
# read-only protocol below is still the only correct object to
# return, so this refusal costs no capability -- what it buys is
# that the request is answered rather than dropped. It used to be
# popped and never consulted here, so the caller learned its
# protocol was read-only from an absent write method somewhere
# downstream (#528) -- the same silent-wrong-mode shape #514
# fixed for the memo rebuild.
#
# Note the mode arm of that rebuild refusal cannot cover this: it
# compares the mode of the class this function *picked*, which on
# this branch is read-only whatever the caller asked for.
raise ValueError(
f"protocol {proto_id!r} over {strip_url_userinfo(root_url)} "
f"cannot be built read-write: an http(s) repository is "
f"read-only -- there is nothing to create over http and no "
f"lockfile to take. Omit read_only to build the read-only "
f"protocol, or use a file:// or s3:// url for a repository "
f"that can be written to")
return FsspecReadOnlyProtocol(
proto_id, root_url,
filesystem=filesystem,
public_url=public_url,
**kwargs)
raise NotImplementedError(f"unsupported schema {url.scheme}")