dandi.utils#

class dandi.utils.Hasher(*args, **kwargs)[source]#
hexdigest() str[source]#
update(data: bytes) None[source]#
dandi.utils.Parallel(**kwargs: Any) Any[source]#

Adapter for joblib.Parallel so we could if desired, centralize control

class dandi.utils.ServerInfo(*, version: str, services: ServerServices, cli_minimal_version: str, cli_bad_versions: List[str])[source]#
cli_bad_versions: List[str]#
cli_minimal_version: str#
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'cli_bad_versions': FieldInfo(annotation=List[str], required=True, alias='cli-bad-versions', alias_priority=2), 'cli_minimal_version': FieldInfo(annotation=str, required=True, alias='cli-minimal-version', alias_priority=2), 'services': FieldInfo(annotation=ServerServices, required=True), 'version': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

services: ServerServices#
version: str#
class dandi.utils.ServerServices(*, api: ServiceURL, webui: ServiceURL | None = None, jupyterhub: ServiceURL | None = None)[source]#
api: ServiceURL#
jupyterhub: ServiceURL | None#
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'api': FieldInfo(annotation=ServiceURL, required=True), 'jupyterhub': FieldInfo(annotation=Union[ServiceURL, NoneType], required=False, default=None), 'webui': FieldInfo(annotation=Union[ServiceURL, NoneType], required=False, default=None)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

webui: ServiceURL | None#
class dandi.utils.ServiceURL(*, url: str)[source]#
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'url': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

url: str#
dandi.utils.abbrev_prompt(msg: str, *options: str) str[source]#

Prompt the user to input one of several options, which can be entered as either a whole word or the first letter of a word. All input is handled case-insensitively. Returns the complete word corresponding to the input, lowercased.

For example, abbrev_prompt("Delete assets?", "yes", "no", "list") prompts the user with the message Delete assets? ([y]es/[n]o/[l]ist): `` and accepts as input ``y`, ``yes, n, no, l, and list.

dandi.utils.check_dandi_version() None[source]#
dandi.utils.chunked(iterable: Iterable[T], size: int) Iterator[list[T]][source]#
dandi.utils.copy_file(src: str | Path, dst: str | Path) None[source]#

Copy file from src to dst

dandi.utils.delayed(*args, **kwargs)[source]#

Adapter for joblib.delayed so we could if desired, centralize control

dandi.utils.ensure_datetime(t: datetime.datetime | int | float | str, strip_tzinfo: bool = False, tz: datetime.tzinfo | None = None) datetime.datetime[source]#

Ensures that time is a datetime

strip_tzinfo applies only to str records passed in

epoch time assumed to be local (not utc)

dandi.utils.ensure_strtime(t: str | int | float | datetime.datetime, isoformat: bool = True) str[source]#

Ensures that time is a string in iso format

Note: if t is already a string, no conversion of any kind is done.

epoch time assumed to be local (not utc)

Parameters:

isoformat (bool, optional) – If True, use .isoformat() and otherwise str(). With .isoformat() there is no space but T to separate date from time.

dandi.utils.exclude_from_zarr(path: Path) bool[source]#

Returns True if the path is a file or directory that should be excluded from consideration when located in a Zarr

dandi.utils.find_files(regex: str, paths: AnyPath | Iterable[AnyPath] = '.', exclude: str | None = None, exclude_dotfiles: bool = True, exclude_dotdirs: bool = True, exclude_vcs: bool = True, exclude_datalad: bool = False, dirs: bool = False, dirs_avoid: str | None = None) Iterator[str][source]#

Generator to find files matching regex

Parameters:
  • regex (string) – Regex to search target files. Is not applied to filter out directories

  • paths (string or list, optional) – Directories or files to search among (directories are searched recursively)

  • exclude (string, optional) – Matches to exclude

  • exclude_vcs – If True, excludes commonly known VCS subdirectories. If string, used as regex to exclude those files (regex: %r)

  • exclude_dotdirs – If True, does not descend into directories starting with “.”.

  • exclude_datalad – If True, excludes files known to be datalad meta-data files (e.g. under .datalad/ subdirectory) (regex: %r)

  • dirs (bool, optional) – Whether to match directories as well as files

  • dirs_avoid (string, optional) – Regex for directories to not rercurse under (they might still be reported if dirs=True)

dandi.utils.find_parent_directory_containing(filename: AnyPath, path: AnyPath | None = None) Path | None[source]#

Find a directory, on the path to ‘path’ containing filename

if no ‘path’ - path from cwd. If ‘path’ is not absolute, absolute path is taken assuming relative to cwd.

Returns None if no such found, pathlib’s Path (absolute) to the directory if found.

dandi.utils.flatten(it: Iterable) Iterator[source]#

Yield items flattened if list, tuple or a generator

dandi.utils.flattened(it: Iterable) list[source]#

Return list with items flattened if list, tuple or a generator

dandi.utils.fromisoformat(t: str) datetime[source]#
dandi.utils.get_instance(dandi_instance_id: str | DandiInstance) DandiInstance[source]#
dandi.utils.get_mime_type(filename: str, strict: bool = False) str[source]#

Like mimetypes.guess_type(), except that if the file is compressed, the MIME type for the compression is returned. Also, the default return value is now 'application/octet-stream' instead of None.

dandi.utils.get_module_version(module: str | types.ModuleType) str | None[source]#

Return version of the module

Return module’s __version__ if present, or use importlib to get version.

Return type:

object

dandi.utils.get_utcnow_datetime(microseconds: bool = True) datetime[source]#

Return current time as datetime with time zone information.

Microseconds are stripped away.

If string representation is desired, just “apply” .isoformat()

dandi.utils.is_interactive() bool[source]#

Return True if all in/outs are tty

dandi.utils.is_page2_url(page1: str, page2: str) bool[source]#

Tests whether the URL page2 is the same as page1 but with the page query parameter set to 2

dandi.utils.is_same_time(*times: datetime.datetime | int | float | str, tolerance: float = 1e-06, strip_tzinfo: bool = False) bool[source]#

Helper to do comparison between time points

Time zone information gets stripped Does it by first normalizing all times to datetime, and then comparing to the first entry

Parameters:

tolerance (float, optional) – Seconds of difference between times to tolerate. By default difference up to a microsecond is ok

dandi.utils.is_url(s: str) bool[source]#

Very primitive url detection for now

TODO: redo

dandi.utils.joinurl(base: str, path: str) str[source]#

Append a slash-separated path to a base HTTP(S) URL base. The two components are separated by a single slash, removing any excess slashes that would be present after naïve concatenation.

If path is already an absolute HTTP(S) URL, it is returned unchanged.

Note that this function differs from urllib.parse.urljoin() when the path portion of base is nonempty and does not end in a slash.

dandi.utils.list_paths(dirpath: AnyPath, dirs: bool = False, exclude_vcs: bool = True) list[Path][source]#
dandi.utils.load_jsonl(filename: str | Path) list[source]#

Load json lines formatted file

dandi.utils.move_file(src: str | Path, dst: str | Path) Any[source]#

Move file from src to dst

dandi.utils.path_is_subpath(path: str, prefix: str) bool[source]#

Return True if path is a subpath of prefix

It will return False if path == prefix.

Parameters:
  • path (str) –

  • prefix (str) –

dandi.utils.pluralize(n: int, word: str, plural: str | None = None) str[source]#
dandi.utils.post_upload_size_check(path: Path, pre_check_size: int, erroring: bool) None[source]#
dandi.utils.pre_upload_size_check(path: Path) int[source]#
dandi.utils.setup_exceptionhook(ipython: bool = False) None[source]#

Overloads default sys.excepthook with our exceptionhook handler.

If interactive, our exceptionhook handler will invoke pdb.post_mortem; if not interactive, then invokes default handler.

dandi.utils.shortened_repr(value: Any, length: int = 30) str[source]#
dandi.utils.under_paths(paths: Iterable[str | PurePath], filter_paths: Iterable[str | PurePath]) Iterator[PurePosixPath][source]#

Return all elements of paths (converted to PurePosixPath instances) that are equal to or under/start with one or more paths in filter_paths. The elements of both iterables must be relative & normalized.

Based on get_filtered_paths_ from datalad’s datalad/support/path.py

dandi.utils.with_pathsep(path: str) str[source]#

Little helper to guarantee that path ends with /

dandi.utils.yaml_dump(rec: Any) str[source]#

Consistent dump into yaml

Of primary importance is default_flow_style=False to assure proper formatting on versions of pyyaml before 5.1: https://github.com/yaml/pyyaml/pull/256

dandi.utils.yaml_load(f: str | IO[str], typ: str | None = None) Any[source]#

Load YAML source from a file or string.

Parameters:
  • f (str or IO[str]) – The YAML source to load

  • typ (str, optional) – The value of typ to pass to ruamel.yaml.YAML. May be “rt” (default), “safe”, “unsafe”, or “base”

Returns:

The parsed YAML value

Return type:

Any