fs

mlte/store/artifact/underlying/fs.py

Implementation of local file system artifact store.

FileSystemGroupMappper

Bases: GroupMapper

FS mapper for the group resource.

Source code in mlte/store/user/underlying/fs.py

class FileSystemGroupMappper(GroupMapper):
    """FS mapper for the group resource."""

    GROUPS_FOLDER = "groups"
    """Subfolder for groups."""

    def __init__(self, storage: FileSystemStorage) -> None:
        self.storage = storage.clone()
        """A reference to underlying storage."""

        self.storage.setup_base_path(
            Path(FileSystemUserStore.BASE_USERS_FOLDER, self.GROUPS_FOLDER)
        )
        """Set the subfolder for this resource."""

    def create(self, group: Group, context: Any = None) -> Group:
        self.storage.ensure_resource_does_not_exist(group.name)
        return self._write_group(group)

    def edit(self, group: Group, context: Any = None) -> Group:
        self.storage.ensure_resource_exists(group.name)
        return self._write_group(group)

    def read(self, group_name: str, context: Any = None) -> Group:
        return self._read_group(group_name)

    def list(self, context: Any = None) -> List[str]:
        return self.storage.list_resources()

    def delete(self, group_name: str, context: Any = None) -> Group:
        self.storage.ensure_resource_exists(group_name)
        group = self._read_group(group_name)
        self.storage.delete_resource(group_name)
        return group

    def _read_group(self, group_name: str) -> Group:
        """
        Lazily construct a Group object on read.
        :param group_name: The group name
        :return: The group object
        """
        self.storage.ensure_resource_exists(group_name)
        return Group(**self.storage.read_resource(group_name))

    def _write_group(self, group: Group) -> Group:
        """Writes a Group to storage."""
        self.storage.write_resource(group.name, group.to_json())
        return self._read_group(group.name)

GROUPS_FOLDER = 'groups' class-attribute instance-attribute

Subfolder for groups.

storage = storage.clone() instance-attribute

A reference to underlying storage.

FileSystemPermissionMappper

Bases: PermissionMapper

FS mapper for the permission resource.

Source code in mlte/store/user/underlying/fs.py

class FileSystemPermissionMappper(PermissionMapper):
    """FS mapper for the permission resource."""

    PERMISSIONS_FOLDER = "permissions"
    """Subfolder for permissions."""

    def __init__(self, storage: FileSystemStorage) -> None:
        self.storage = storage.clone()
        """A reference to underlying storage."""

        self.storage.setup_base_path(
            Path(FileSystemUserStore.BASE_USERS_FOLDER, self.PERMISSIONS_FOLDER)
        )
        """Set the subfolder for this resource."""

    def create(self, permission: Permission, context: Any = None) -> Permission:
        self.storage.ensure_resource_does_not_exist(permission.to_str())
        return self._write_permission(permission)

    def edit(self, permission: Permission, context: Any = None) -> Permission:
        self.storage.ensure_resource_exists(permission.to_str())
        return self._write_permission(permission)

    def read(self, permission_str: str, context: Any = None) -> Permission:
        return self._read_permission(permission_str)

    def list(self, context: Any = None) -> List[str]:
        return self.storage.list_resources()

    def delete(self, permission_str: str, context: Any = None) -> Permission:
        self.storage.ensure_resource_exists(permission_str)
        permission = self._read_permission(permission_str)
        self.storage.delete_resource(permission_str)
        return permission

    def _read_permission(self, permission_str: str) -> Permission:
        """
        Lazily construct a Permission object on read.
        :param permission_str: The permission str
        :return: The permission object
        """
        self.storage.ensure_resource_exists(permission_str)
        return Permission(**self.storage.read_resource(permission_str))

    def _write_permission(self, permission: Permission) -> Permission:
        """Writes a Permission to storage."""
        self.storage.write_resource(permission.to_str(), permission.to_json())
        return self._read_permission(permission.to_str())

PERMISSIONS_FOLDER = 'permissions' class-attribute instance-attribute

Subfolder for permissions.

storage = storage.clone() instance-attribute

A reference to underlying storage.

FileSystemUserMappper

Bases: UserMapper

FS mapper for the user resource.

Source code in mlte/store/user/underlying/fs.py

class FileSystemUserMappper(UserMapper):
    """FS mapper for the user resource."""

    USERS_FOLDER = "users"
    """Subfolder for users."""

    def __init__(
        self, storage: FileSystemStorage, group_mapper: FileSystemGroupMappper
    ) -> None:
        self.storage = storage.clone()
        """A reference to underlying storage."""

        self.group_mapper = group_mapper
        """Reference to group mapper, to get updated groups when needed."""

        self.storage.setup_base_path(
            Path(FileSystemUserStore.BASE_USERS_FOLDER, self.USERS_FOLDER)
        )
        """Set the subfolder for this resource."""

    def create(self, user: UserWithPassword, context: Any = None) -> User:
        self.storage.ensure_resource_does_not_exist(user.username)

        new_user = user.to_hashed_user()

        # Only store group names for consistency.
        new_user.groups = Group.get_group_names(new_user.groups)

        return self._write_user(new_user)

    def edit(
        self, user: Union[UserWithPassword, BasicUser], context: Any = None
    ) -> User:
        # NOTE: a JSON file may not have the updated group data, which can make reading the JSON confusing.
        self.storage.ensure_resource_exists(user.username)

        curr_user = self._read_user(user.username)
        updated_user = update_user_data(curr_user, user)

        # Only store group names for consistency.
        updated_user.groups = Group.get_group_names(updated_user.groups)

        return self._write_user(updated_user)

    def read(self, username: str, context: Any = None) -> User:
        user = self._read_user(username)

        # Now get updated info for each group.
        up_to_date_groups: List[Group] = []
        for group in user.groups:
            up_to_date_groups.append(self.group_mapper.read(group.name))
        user.groups = up_to_date_groups

        return user

    def list(self, context: Any = None) -> List[str]:
        return self.storage.list_resources()

    def delete(self, username: str, context: Any = None) -> User:
        self.storage.ensure_resource_exists(username)
        user = self._read_user(username)
        self.storage.delete_resource(username)
        return user

    def _read_user(self, username: str) -> User:
        """
        Lazily construct a User object on read.
        :param username: The username
        :return: The user object
        """
        self.storage.ensure_resource_exists(username)
        return User(**self.storage.read_resource(username))

    def _write_user(self, user: User) -> User:
        """Writes a user to storage."""
        self.storage.write_resource(user.username, user.to_json())
        return user

USERS_FOLDER = 'users' class-attribute instance-attribute

Subfolder for users.

group_mapper = group_mapper instance-attribute

Reference to group mapper, to get updated groups when needed.

storage = storage.clone() instance-attribute

A reference to underlying storage.

FileSystemUserStore

Bases: UserStore

A local file system implementation of the MLTE user store.

Source code in mlte/store/user/underlying/fs.py

class FileSystemUserStore(UserStore):
    """A local file system implementation of the MLTE user store."""

    BASE_USERS_FOLDER = "users"
    """Base folder to store users store in."""

    def __init__(self, uri: StoreURI) -> None:
        self.storage = FileSystemStorage(
            uri=uri, sub_folder=self.BASE_USERS_FOLDER
        )
        """Underlying storage."""

        # Initialize defaults.
        super().__init__(uri=uri)

    def session(self) -> FileSystemUserStoreSession:
        """
        Return a session handle for the store instance.
        :return: The session handle
        """
        return FileSystemUserStoreSession(storage=self.storage)

BASE_USERS_FOLDER = 'users' class-attribute instance-attribute

Base folder to store users store in.

storage = FileSystemStorage(uri=uri, sub_folder=self.BASE_USERS_FOLDER) instance-attribute

Underlying storage.

session()

Return a session handle for the store instance.

Returns:

Type	Description
FileSystemUserStoreSession	The session handle

Source code in mlte/store/user/underlying/fs.py

def session(self) -> FileSystemUserStoreSession:
    """
    Return a session handle for the store instance.
    :return: The session handle
    """
    return FileSystemUserStoreSession(storage=self.storage)

FileSystemUserStoreSession

Bases: UserStoreSession

A local file-system implementation of the MLTE user store.

Source code in mlte/store/user/underlying/fs.py

class FileSystemUserStoreSession(UserStoreSession):
    """A local file-system implementation of the MLTE user store."""

    def __init__(self, storage: FileSystemStorage) -> None:
        self.permission_mapper = FileSystemPermissionMappper(storage)
        """The mapper to permisison CRUD."""

        self.group_mapper = FileSystemGroupMappper(storage)
        """The mapper to group CRUD."""

        self.user_mapper = FileSystemUserMappper(storage, self.group_mapper)
        """The mapper to user CRUD."""

    def close(self) -> None:
        """Close the session."""
        # Closing a local FS session is a no-op.
        pass

group_mapper = FileSystemGroupMappper(storage) instance-attribute

The mapper to group CRUD.

permission_mapper = FileSystemPermissionMappper(storage) instance-attribute

The mapper to permisison CRUD.

user_mapper = FileSystemUserMappper(storage, self.group_mapper) instance-attribute

The mapper to user CRUD.

close()

Close the session.

Source code in mlte/store/user/underlying/fs.py

def close(self) -> None:
    """Close the session."""
    # Closing a local FS session is a no-op.
    pass