locations

locations

Dagster resource for accessing filesystem locations.

Classes

Location

Location(config)

Filesystem plus path.

Source code in niceml/dagster/resources/locations.py

def __init__(self, config: LocationConfig):
    self._config = deepcopy(config)

Functions

__str__

__str__()

Return string representation of config.

Source code in niceml/dagster/resources/locations.py

def __str__(self) -> str:
    """Return string representation of config."""
    return str(self._config)

open

open(rel_path, *args, **kwargs)

Open file at rel_path with *args and **kwargs.

Source code in niceml/dagster/resources/locations.py

@contextmanager
def open(self, rel_path: str, *args, **kwargs) -> Iterator[Any]:
    """Open file at ``rel_path`` with ``*args`` and ``**kwargs``."""
    with self.open_fs_path() as (filesystem, path):
        absolute_path = os.path.join(path, rel_path)
        yield filesystem.open(absolute_path, *args, **kwargs)

open_fs_path

open_fs_path()

Return filesystem and path as context manager.

Source code in niceml/dagster/resources/locations.py

@contextmanager
def open_fs_path(self) -> Iterator[FSPath]:
    """Return filesystem and path as context manager."""
    with open_location(asdict(self._config)) as (filesystem, path):
        yield (filesystem, path)

LocationsResource

LocationsResource(config)

Resource for accessing file system.

Parameters:

• config (LocationsResourceConfig) –

  resource configuration.

Source code in niceml/dagster/resources/locations.py

def __init__(self, config: LocationsResourceConfig):
    location_items = config.locations.items()
    self._locations = {key: Location(value) for key, value in location_items}

Functions

__getitem__

__getitem__(location_name)

Return configured location.

Source code in niceml/dagster/resources/locations.py

def __getitem__(self, location_name: str) -> Location:
    """Return configured location."""
    try:
        location = self._locations[location_name]
    except KeyError as exception:
        getLogger(__name__).error("location %s not configured", location_name)
        raise exception
    getLogger(__name__).debug("opening location %s (%s)", location_name, location)
    return location

LocationsResourceConfig

Configuration of file system locations.

Functions

get_location

get_location(location_name)

Return location.

Source code in niceml/dagster/resources/locations.py

def get_location(location_name: str) -> Location:
    """Return location."""
    if _LOCATIONS_RESOURCE is None:
        logging.error(
            """Locations resource not configured. Define the resource
            ``global_fs_resource`` for your dagster job and require it for your op."""
        )
        raise ValueError("Locations resource not configured. See log for details.")
    return _LOCATIONS_RESOURCE[location_name]

global_locations_resource

global_locations_resource(context)

Return LocationsResource and register it as module-level object FS.

Source code in niceml/dagster/resources/locations.py

@resource
def global_locations_resource(context: InitResourceContext):
    """Return `LocationsResource` and register it as module-level object ``FS``."""
    global _LOCATIONS_RESOURCE  # pylint: disable=global-statement
    _LOCATIONS_RESOURCE = locations_resource(context)
    return _LOCATIONS_RESOURCE

locations_resource

locations_resource(context)

Create LocationsResource according to the configuration.

Source code in niceml/dagster/resources/locations.py

@resource(config_schema=build_config_schema(LocationsResourceConfig))
def locations_resource(context: InitResourceContext):
    """Create LocationsResource according to the configuration."""
    config = parse_config(context.resource_config, LocationsResourceConfig)
    return LocationsResource(config)