_gdrive
sleap_io.io._gdrive
¶
Google Drive share-link resolution for remote loading.
Google Drive share URLs (https://drive.google.com/file/d/<ID>/view,
https://drive.google.com/uc?id=<ID>, …/open?id=<ID>) do not point at the
file bytes directly: Drive serves an HTML "interstitial" confirmation page for
anything large enough to skip its virus scan, and the real download URL lives
inside that page's #download-form (or, for small files, behind a
Content-Disposition header on the first response). This module ports the
minimal subset of gdown <https://github.com/wkentaro/gdown>_'s resolution
logic needed by sleap-io, using only the standard library
(:mod:html.parser + :mod:re) for HTML parsing -- no gdown or
beautifulsoup4 dependency.
The resolver (:func:_open_gdrive) is notebook-safe: it performs its HTTP
GETs on fsspec's background event-loop thread via :func:fsspec.asyn.sync, so
it does not call :func:asyncio.run (which raises "event loop is already
running" inside a Jupyter kernel). Because Drive rejects HEAD with 405
(which breaks fsspec's lazy size-probe) and its download quota can interrupt a
transfer mid-stream, the resolver fully prefetches the resolved bytes within a
single cookie-carrying session and returns an :class:io.BytesIO.
__all__ = ['_GDRIVE_HOSTS', '_is_gdrive_url']
module-attribute
¶
Built-in mutable sequence.
If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.
__cached__ = '/home/runner/work/sleap-io/sleap-io/sleap_io/io/__pycache__/_gdrive.cpython-313.pyc'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__doc__ = 'Google Drive share-link resolution for remote loading.\n\nGoogle Drive share URLs (``https://drive.google.com/file/d/<ID>/view``,\n``https://drive.google.com/uc?id=<ID>``, ``…/open?id=<ID>``) do not point at the\nfile bytes directly: Drive serves an HTML "interstitial" confirmation page for\nanything large enough to skip its virus scan, and the real download URL lives\ninside that page\'s ``#download-form`` (or, for small files, behind a\n``Content-Disposition`` header on the first response). This module ports the\nminimal subset of `gdown <https://github.com/wkentaro/gdown>`_\'s resolution\nlogic needed by sleap-io, using only the standard library\n(:mod:`html.parser` + :mod:`re`) for HTML parsing -- no ``gdown`` or\n``beautifulsoup4`` dependency.\n\nThe resolver (:func:`_open_gdrive`) is **notebook-safe**: it performs its HTTP\nGETs on fsspec\'s background event-loop thread via :func:`fsspec.asyn.sync`, so\nit does not call :func:`asyncio.run` (which raises "event loop is already\nrunning" inside a Jupyter kernel). Because Drive rejects ``HEAD`` with 405\n(which breaks fsspec\'s lazy size-probe) and its download quota can interrupt a\ntransfer mid-stream, the resolver fully prefetches the resolved bytes within a\nsingle cookie-carrying session and returns an :class:`io.BytesIO`.\n'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__file__ = '/home/runner/work/sleap-io/sleap-io/sleap_io/io/_gdrive.py'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__name__ = 'sleap_io.io._gdrive'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__package__ = 'sleap_io.io'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
RemoteIOError
¶
Bases: builtins.OSError
Raised for HTTP-level failures during remote loading.
Subclasses :class:OSError so that callers which already handle
OSError (e.g. HDF5Video.__attrs_post_init__) degrade gracefully.
The original aiohttp/fsspec exception is intentionally not chained as
__cause__ / __context__ because its string form embeds the raw URL
(userinfo password, ?token= query parameter, etc.), which
:func:traceback.format_exception would otherwise leak. A redacted summary
of the cause (type name + redacted message) is preserved instead via
:attr:cause_summary.
Attributes:
| Name | Type | Description |
|---|---|---|
url |
Redacted URL (credentials stripped if present), or None. |
|
status |
HTTP status code, or None for connection-level errors. |
|
cause_summary |
Redacted, one-line summary of the underlying exception (type name plus redacted message), or None. |
Methods:
| Name | Description |
|---|---|
__init__ |
Build a RemoteIOError with a redacted, composed message. |
Source code in sleap_io/io/_remote.py
class RemoteIOError(OSError):
"""Raised for HTTP-level failures during remote loading.
Subclasses :class:`OSError` so that callers which already handle
``OSError`` (e.g. ``HDF5Video.__attrs_post_init__``) degrade gracefully.
The original aiohttp/fsspec exception is intentionally *not* chained as
``__cause__`` / ``__context__`` because its string form embeds the raw URL
(userinfo password, ``?token=`` query parameter, etc.), which
:func:`traceback.format_exception` would otherwise leak. A redacted summary
of the cause (type name + redacted message) is preserved instead via
:attr:`cause_summary`.
Attributes:
url: Redacted URL (credentials stripped if present), or None.
status: HTTP status code, or None for connection-level errors.
cause_summary: Redacted, one-line summary of the underlying exception
(type name plus redacted message), or None.
"""
def __init__(
self,
message: str,
*,
url: str | None = None,
status: int | None = None,
cause_summary: str | None = None,
) -> None:
"""Build a RemoteIOError with a redacted, composed message.
Args:
message: Human-readable description of the failure.
url: Raw URL associated with the failure. It is redacted before
being stored or surfaced in the message.
status: HTTP status code, if known.
cause_summary: A pre-redacted summary of the underlying exception
(see :func:`_redacted_cause_summary`). It is stored on the
instance and appended to the message. The raw exception is
never chained, so no credential-bearing string can leak through
``__cause__`` / ``__context__``.
"""
self.url = _redact_url(url) if url else None
self.status = status
self.cause_summary = cause_summary
parts = [message]
if self.status is not None:
parts.append(f"status={self.status}")
if self.url:
parts.append(f"url={self.url}")
if self.cause_summary:
parts.append(f"cause={self.cause_summary}")
super().__init__("; ".join(parts))
__doc__ = 'Raised for HTTP-level failures during remote loading.\n\nSubclasses :class:`OSError` so that callers which already handle\n``OSError`` (e.g. ``HDF5Video.__attrs_post_init__``) degrade gracefully.\n\nThe original aiohttp/fsspec exception is intentionally *not* chained as\n``__cause__`` / ``__context__`` because its string form embeds the raw URL\n(userinfo password, ``?token=`` query parameter, etc.), which\n:func:`traceback.format_exception` would otherwise leak. A redacted summary\nof the cause (type name + redacted message) is preserved instead via\n:attr:`cause_summary`.\n\nAttributes:\n url: Redacted URL (credentials stripped if present), or None.\n status: HTTP status code, or None for connection-level errors.\n cause_summary: Redacted, one-line summary of the underlying exception\n (type name plus redacted message), or None.\n'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__firstlineno__ = 113
class-attribute
¶
int([x]) -> integer int(x, base=10) -> integer
Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating-point numbers, this truncates towards zero.
If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer iteral.
int('0b100', base=0) 4
__module__ = 'sleap_io.io._remote'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__static_attributes__ = ('cause_summary', 'status', 'url')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__weakref__
property
¶
list of weak references to the object
__init__(message, *, url=None, status=None, cause_summary=None)
¶
Build a RemoteIOError with a redacted, composed message.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Human-readable description of the failure. |
required |
url
|
str | None
|
Raw URL associated with the failure. It is redacted before being stored or surfaced in the message. |
None
|
status
|
int | None
|
HTTP status code, if known. |
None
|
cause_summary
|
str | None
|
A pre-redacted summary of the underlying exception
(see :func: |
None
|
Source code in sleap_io/io/_remote.py
def __init__(
self,
message: str,
*,
url: str | None = None,
status: int | None = None,
cause_summary: str | None = None,
) -> None:
"""Build a RemoteIOError with a redacted, composed message.
Args:
message: Human-readable description of the failure.
url: Raw URL associated with the failure. It is redacted before
being stored or surfaced in the message.
status: HTTP status code, if known.
cause_summary: A pre-redacted summary of the underlying exception
(see :func:`_redacted_cause_summary`). It is stored on the
instance and appended to the message. The raw exception is
never chained, so no credential-bearing string can leak through
``__cause__`` / ``__context__``.
"""
self.url = _redact_url(url) if url else None
self.status = status
self.cause_summary = cause_summary
parts = [message]
if self.status is not None:
parts.append(f"status={self.status}")
if self.url:
parts.append(f"url={self.url}")
if self.cause_summary:
parts.append(f"cause={self.cause_summary}")
super().__init__("; ".join(parts))