Skip to content

Snirk API Docs :: fs

snirk.fs

Snirk API for connecting to and interacting with SNI device filesystems.

SnirkFilesystem

Bases: Snirk

Container and API for connecting to and interacting with SNI devices with access to device filesystem.

Each operation assures that the SNI device has necessary capabilities before attempting to access the device filesystem, raising a SnirkIncapableError exception when the capabilities are missing.

Parameters:

Name Type Description Default
max_message_length int

max bytes to transfer in get/put operations

 required
Source code in snirk/fs.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class SnirkFilesystem(Snirk):
    """
    Container and API for connecting to and interacting with SNI devices with access to device filesystem.

    Each operation assures that the SNI device has necessary capabilities before attempting to access
    the device filesystem, raising a SnirkIncapableError exception when the capabilities are missing.

    :param max_message_length: max bytes to transfer in get/put operations
    :type max_message_length: int
    """

    max_message_length: int = 1024**3

    async def get_file(self, device_file: str, timeout: int = 30) -> pb.GetFileResponse:
        """Get device file from SNI device filesystem.

        :param device_file: path to device file to read
        :param timeout: seconds until timeout reading file from device
        :raises SnirkIncapableError: when device does not have GetFile capability
        """
        device = await self.find_device()
        await self.ensure_capability(pb.DeviceCapability.GetFile)

        options = [
            ("grpc.max_message_length", self.max_message_length),
            ("grpc.max_receive_message_length", self.max_message_length),
        ]
        async with self.grpc_channel(options=options, timeout=timeout) as ch:
            stub = sni.DeviceFilesystemStub(ch)
            retry_get = await self.retry_grpc(stub.GetFile, max_retries=3, sleep=0.5)
            return await retry_get(pb.GetFileRequest(uri=device.uri, path=device_file))

    async def make_directory(self, device_path: str, parents: bool = False, timeout: int = 10):
        """Create missing directories on SNI device filesystem.

        If parents is not true, then an exception will propagate from gRPC if device_path
        cannot be created because it's parent (sub-)directories do not exist. When parents is
        true, all missing (sub-)directories will be created in the correct order.

        :param device_path: path to create on SNI device
        :param parents: when True, create any missing parent (sub-)directories for device_path
        :param timeout: seconds to try creating directories until timeout
        :raises SnirkIncapableError: when device does not have MakeDirectory capability
        """
        directories: list[str] = [device_path]
        if parents:
            parts = Path(device_path).parts
            directories = [str(Path(*parts[0:x])) for x in range(2, len(parts) + 1)]

        device = await self.find_device()
        await self.ensure_capability(pb.DeviceCapability.MakeDirectory)

        # find and create missing directories
        async with self.grpc_channel(timeout=timeout) as ch:
            stub = sni.DeviceFilesystemStub(ch)
            for directory in directories:
                try:
                    stub.ReadDirectory(pb.ReadDirectoryRequest(uri=device.uri, path=directory))
                except grpc._channel._InactiveRpcError:
                    retry_mkdir = await self.retry_grpc(stub.MakeDirectory)
                    await retry_mkdir(pb.MakeDirectoryRequest(uri=device.uri, path=directory))

    async def put_file(self, data: bytes, device_path: str, timeout: int = 30) -> pb.PutFileResponse:
        """Put file bytes at path on SNI device filesystem.

        :param data: bytes to write to device path
        :param device_path: path on device to write local_file bytes
        :param timeout: seconds until timeout writing file to device
        :raises SnirkIncapableError: when device does not have PutFile capability
        """
        device = await self.find_device()
        await self.ensure_capability(pb.DeviceCapability.PutFile)

        options = [
            ("grpc.max_message_length", self.max_message_length),
            ("grpc.max_send_message_length", self.max_message_length),
        ]
        async with self.grpc_channel(options=options, timeout=timeout) as ch:
            stub = sni.DeviceFilesystemStub(ch)
            retry_put = await self.retry_grpc(stub.PutFile, max_retries=3, sleep=0.5)
            response = await retry_put(pb.PutFileRequest(uri=device.uri, path=device_path, data=data))

        # note: appears that PutFileResponse.size is missing
        # might be SNI bug or could be gRPC proto error (it specifies size)
        return response

    async def read_directory(self, path: str, timeout: int = 2) -> pb.ReadDirectoryResponse:
        """Read directory from SNI device filesystem.

        :param path: directory path to read from device
        :param timeout: seconds until timeout reading directory from device
        :raises SnirkIncapableError: when device does not have ReadDirectory capability
        """
        device = await self.find_device()
        await self.ensure_capability(pb.DeviceCapability.ReadDirectory)

        async with self.grpc_channel() as ch:
            stub = sni.DeviceFilesystemStub(ch)
            retry = await self.retry_grpc(stub.ReadDirectory, timeout=timeout)
            try:
                response = await retry(pb.ReadDirectoryRequest(uri=device.uri, path=path))
            except grpc._channel._InactiveRpcError as ex:
                if "failed to list" not in str(ex):
                    raise
                # otherwise, could be because path is a single file, or because doesn't exist
                # if doesn't exist, let exception propagate up
                parent = str(Path(path).parent)
                response = await retry(pb.ReadDirectoryRequest(uri=device.uri, path=parent))
        return response

    async def remove_file(self, device_file: str, timeout: int = 5) -> pb.RemoveFileResponse:
        """Remove device file from SNI device filesystem.

        :param device_file: path to device file to remove
        :param timeout: seconds until timeout removing file from device
        :raises SnirkIncapableError: when device does not have RemoveFile capability
        """
        device = await self.find_device()
        await self.ensure_capability(pb.DeviceCapability.RemoveFile)

        async with self.grpc_channel(timeout=timeout) as ch:
            stub = sni.DeviceFilesystemStub(ch)
            retry_get = await self.retry_grpc(stub.RemoveFile, max_retries=3, sleep=0.5)
            return await retry_get(pb.RemoveFileRequest(uri=device.uri, path=device_file))

get_file async 

get_file(device_file, timeout=30)

Get device file from SNI device filesystem.

Parameters:

Name Type Description Default
device_file str

path to device file to read

 required
timeout int

seconds until timeout reading file from device

 30

Raises:

Type Description
SnirkIncapableError

when device does not have GetFile capability
Source code in snirk/fs.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
async def get_file(self, device_file: str, timeout: int = 30) -> pb.GetFileResponse:
    """Get device file from SNI device filesystem.

    :param device_file: path to device file to read
    :param timeout: seconds until timeout reading file from device
    :raises SnirkIncapableError: when device does not have GetFile capability
    """
    device = await self.find_device()
    await self.ensure_capability(pb.DeviceCapability.GetFile)

    options = [
        ("grpc.max_message_length", self.max_message_length),
        ("grpc.max_receive_message_length", self.max_message_length),
    ]
    async with self.grpc_channel(options=options, timeout=timeout) as ch:
        stub = sni.DeviceFilesystemStub(ch)
        retry_get = await self.retry_grpc(stub.GetFile, max_retries=3, sleep=0.5)
        return await retry_get(pb.GetFileRequest(uri=device.uri, path=device_file))

make_directory async 

make_directory(device_path, parents=False, timeout=10)

Create missing directories on SNI device filesystem.

If parents is not true, then an exception will propagate from gRPC if device_path cannot be created because it's parent (sub-)directories do not exist. When parents is true, all missing (sub-)directories will be created in the correct order.

Parameters:

Name Type Description Default
device_path str

path to create on SNI device

 required
parents bool

when True, create any missing parent (sub-)directories for device_path

 False
timeout int

seconds to try creating directories until timeout

 10

Raises:

Type Description
SnirkIncapableError

when device does not have MakeDirectory capability
Source code in snirk/fs.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
async def make_directory(self, device_path: str, parents: bool = False, timeout: int = 10):
    """Create missing directories on SNI device filesystem.

    If parents is not true, then an exception will propagate from gRPC if device_path
    cannot be created because it's parent (sub-)directories do not exist. When parents is
    true, all missing (sub-)directories will be created in the correct order.

    :param device_path: path to create on SNI device
    :param parents: when True, create any missing parent (sub-)directories for device_path
    :param timeout: seconds to try creating directories until timeout
    :raises SnirkIncapableError: when device does not have MakeDirectory capability
    """
    directories: list[str] = [device_path]
    if parents:
        parts = Path(device_path).parts
        directories = [str(Path(*parts[0:x])) for x in range(2, len(parts) + 1)]

    device = await self.find_device()
    await self.ensure_capability(pb.DeviceCapability.MakeDirectory)

    # find and create missing directories
    async with self.grpc_channel(timeout=timeout) as ch:
        stub = sni.DeviceFilesystemStub(ch)
        for directory in directories:
            try:
                stub.ReadDirectory(pb.ReadDirectoryRequest(uri=device.uri, path=directory))
            except grpc._channel._InactiveRpcError:
                retry_mkdir = await self.retry_grpc(stub.MakeDirectory)
                await retry_mkdir(pb.MakeDirectoryRequest(uri=device.uri, path=directory))

put_file async 

put_file(data, device_path, timeout=30)

Put file bytes at path on SNI device filesystem.

Parameters:

Name Type Description Default
data bytes

bytes to write to device path

 required
device_path str

path on device to write local_file bytes

 required
timeout int

seconds until timeout writing file to device

 30

Raises:

Type Description
SnirkIncapableError

when device does not have PutFile capability
Source code in snirk/fs.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def put_file(self, data: bytes, device_path: str, timeout: int = 30) -> pb.PutFileResponse:
    """Put file bytes at path on SNI device filesystem.

    :param data: bytes to write to device path
    :param device_path: path on device to write local_file bytes
    :param timeout: seconds until timeout writing file to device
    :raises SnirkIncapableError: when device does not have PutFile capability
    """
    device = await self.find_device()
    await self.ensure_capability(pb.DeviceCapability.PutFile)

    options = [
        ("grpc.max_message_length", self.max_message_length),
        ("grpc.max_send_message_length", self.max_message_length),
    ]
    async with self.grpc_channel(options=options, timeout=timeout) as ch:
        stub = sni.DeviceFilesystemStub(ch)
        retry_put = await self.retry_grpc(stub.PutFile, max_retries=3, sleep=0.5)
        response = await retry_put(pb.PutFileRequest(uri=device.uri, path=device_path, data=data))

    # note: appears that PutFileResponse.size is missing
    # might be SNI bug or could be gRPC proto error (it specifies size)
    return response

read_directory async 

read_directory(path, timeout=2)

Read directory from SNI device filesystem.

Parameters:

Name Type Description Default
path str

directory path to read from device

 required
timeout int

seconds until timeout reading directory from device

 2

Raises:

Type Description
SnirkIncapableError

when device does not have ReadDirectory capability
Source code in snirk/fs.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def read_directory(self, path: str, timeout: int = 2) -> pb.ReadDirectoryResponse:
    """Read directory from SNI device filesystem.

    :param path: directory path to read from device
    :param timeout: seconds until timeout reading directory from device
    :raises SnirkIncapableError: when device does not have ReadDirectory capability
    """
    device = await self.find_device()
    await self.ensure_capability(pb.DeviceCapability.ReadDirectory)

    async with self.grpc_channel() as ch:
        stub = sni.DeviceFilesystemStub(ch)
        retry = await self.retry_grpc(stub.ReadDirectory, timeout=timeout)
        try:
            response = await retry(pb.ReadDirectoryRequest(uri=device.uri, path=path))
        except grpc._channel._InactiveRpcError as ex:
            if "failed to list" not in str(ex):
                raise
            # otherwise, could be because path is a single file, or because doesn't exist
            # if doesn't exist, let exception propagate up
            parent = str(Path(path).parent)
            response = await retry(pb.ReadDirectoryRequest(uri=device.uri, path=parent))
    return response

remove_file async 

remove_file(device_file, timeout=5)

Remove device file from SNI device filesystem.

Parameters:

Name Type Description Default
device_file str

path to device file to remove

 required
timeout int

seconds until timeout removing file from device

 5

Raises:

Type Description
SnirkIncapableError

when device does not have RemoveFile capability
Source code in snirk/fs.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
async def remove_file(self, device_file: str, timeout: int = 5) -> pb.RemoveFileResponse:
    """Remove device file from SNI device filesystem.

    :param device_file: path to device file to remove
    :param timeout: seconds until timeout removing file from device
    :raises SnirkIncapableError: when device does not have RemoveFile capability
    """
    device = await self.find_device()
    await self.ensure_capability(pb.DeviceCapability.RemoveFile)

    async with self.grpc_channel(timeout=timeout) as ch:
        stub = sni.DeviceFilesystemStub(ch)
        retry_get = await self.retry_grpc(stub.RemoveFile, max_retries=3, sleep=0.5)
        return await retry_get(pb.RemoveFileRequest(uri=device.uri, path=device_file))