API Reference

Utilities available for import directly from the file_keeper module.

make_storage(name, settings)

Initialize storage instance with specified settings.

Storage adapter is defined by type key of the settings. The rest of settings depends on the specific adapter.

Example

storage = make_storage("memo", {"type": "file_keeper:memory"})

PARAMETER	DESCRIPTION
name	name of the storage TYPE: str
settings	configuration for the storage TYPE: dict[str, Any]

RETURNS	DESCRIPTION
Storage	storage instance

RAISES	DESCRIPTION
UnknownAdapterError	storage adapter is not registered

get_storage(name, settings=None)

Get storage from the pool.

If storage accessed for the first time, it's initialized and added to the pool. After that the same storage is returned every time the function is called with the given name.

Settings are required only for initialization, so you can omit them if you are sure that storage exists. Additionally, if settins are not specified but storage is missing from the pool, file-keeper makes an attempt to initialize storage using global configuration. Global configuration can be provided as:

• FILE_KEEPER_CONFIG environment variable that points to a file with configuration
• .file-keeper.json in the current directory hierarchy
• file-keeper/file-keeper.json in the user's config directory(usually, ~/.config/) when platformdirs installed in the environment, for example via pip install 'file-keeper[user_config]' extras.

File must contain storage configuration provided in format

{
    "storages": {
        "my_storage": {  # (1)!
            "type": "file_keeper:memory"  # (2)!
        }
    }
}

1. Name of the storage
2. Options for the storage

JSON configuration is used by default, because python has built-in JSON support. Additional file extensions are checked if environment contains correspoinding package:

Package	Extension
tomllib	.toml
tomli	.toml
pyyaml	.yaml, .yml

Extensions are checked in order .toml, .yaml, .yml, .json.

Example

If storage accessed for the first time, settings are required

>>> storage = get_storage("memory", {"type": "file_keeper:memory"})
>>> storage
<file_keeper.default.adapters.memory.MemoryStorage object at 0x...>

and the same storage is returned every time in subsequent calls

>>> cached = get_storage("memory")
>>> storage is cached
True

but if storage does not exist and settings are omitted, exception is raised

>>> get_storage("new-memory")
Traceback (most recent call last):
    ...
file_keeper.core.exceptions.UnknownStorageError: Storage new-memory is not configured

PARAMETER	DESCRIPTION
name	name of the storage TYPE: str
settings	configuration for the storage TYPE: dict[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
Storage	storage instance

RAISES	DESCRIPTION
UnknownStorageError	storage with the given name is not configured

make_upload(value)

Convert value into Upload object.

Use this function for simple and reliable initialization of Upload object. Avoid creating Upload manually, unless you are 100% sure you can provide correct MIMEtype, size and stream.

Example

Bytes, binary streams, file objects, and a number of other types can be converted into Upload object.

>>> upload = make_upload(b"hello world")

Upload object contains generic information about the file.

>>> upload.size
11
>>> upload.content_type
'text/plain'

Unsupported types will raise TypeError.

>>> make_upload("unicode string")
Traceback (most recent call last):
  ...
TypeError: <class 'str'> cannot be converted into Upload

PARAMETER	DESCRIPTION
value	content of the file TYPE: Any

RAISES	DESCRIPTION
TypeError	if value cannot be converted into Upload object

RETURNS	DESCRIPTION
Upload	upload object with specified content

Storage

Base class for storage implementation.

PARAMETER	DESCRIPTION
settings	storage configuration TYPE: Mapping[str, Any] | Settings

Example

Extend base class to implement custom storage

class MyStorage(Storage):
    SettingsFactory = Settings
    UploaderFactory = Uploader
    ManagerFactory = Manager
    ReaderFactory = Reader

then initialize it using required settings

my_storage = MyStorage({"option": "value"})

SettingsFactory = Settings class-attribute

Factory class for storage settings.

UploaderFactory = Uploader class-attribute

Factory class for uploader service.

ManagerFactory = Manager class-attribute

Factory class for manager service.

ReaderFactory = Reader class-attribute

Factory class for reader service.

capabilities = self.compute_capabilities() class-attribute instance-attribute

Operations supported by storage. Computed from capabilities of services during storage initialization.

hidden = False class-attribute instance-attribute

Flag that marks unsafe/experimental storages.

make_uploader()

Initialize uploader service.

make_manager()

Initialize manager service.

make_reader()

Initialize reader service.

configure(settings) classmethod

Initialize storage configuration.

This method is responsible for transforming mapping with options into storage's settings. It also can initialize additional services and perform extra work, like verifying that storage is ready to accept uploads.

PARAMETER	DESCRIPTION
settings	mapping with storage configuration TYPE: Mapping[str, Any] | Settings

RETURNS	DESCRIPTION
Settings	initialized settings object

compute_capabilities()

Computes the capabilities of the storage based on its services.

Combines the capabilities of the uploader, manager, and reader services, then excludes any capabilities that are listed in the storage settings as disabled.

RETURNS	DESCRIPTION
Capability	The combined capabilities of the storage.

supports(operation)

Check whether the storage supports operation.

PARAMETER	DESCRIPTION
operation	capability to check TYPE: Capability

RETURNS	DESCRIPTION
bool	True if operation is supported, False otherwise

supports_synthetic(operation, dest)

Check if the storage can emulate operation using other operations.

This method checks if the storage can perform the specified operation using a combination of other supported operations, often in conjunction with a destination storage.

Synthetic operations are not stable and may change in future. They are not considered when computing capabilities of the storage. There are two main reasons to use them:

• required operation involves two storage. For example, moving or copying file from one storage to another.
• operation is not natively supported by the storage, but can be emulated using other operations. For example, RANGE capability means that storage can return specified slice of the file. When this capability is not natively supported, storage can use STREAM capability to read the whole file, returning only specified fragment. This is not efficient but still can solve certain problems.

PARAMETER	DESCRIPTION
operation	capability to check TYPE: Capability
dest	destination storage for operations that require it TYPE: Storage

RETURNS	DESCRIPTION
bool	True if operation is supported, False otherwise

full_path(location, /, **kwargs)

Compute path to the file from the storage's root.

This method works as a shortcut for enabling path option. Whenever your custom storage works with location provided by user, wrap this location into this method to get full path:

class MyCustomReader:
    def stream(self, data: FileData, extras):
        real_location = self.storage.full_path(data_location)
        ...

PARAMETER	DESCRIPTION
location	location of the file object TYPE: Location
**kwargs	exra parameters for custom storages TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
str	full path required to access location

RAISES	DESCRIPTION
LocationError	when location is outside of the storage's path

prepare_location(location, sample=None, /, **kwargs)

Transform and sanitize location using configured functions.

This method applies all transformations configured in location_transformers setting to the provided location. Each transformer is called in the order they are listed in the setting. The output of the previous transformer is passed as an input to the next one.

Example

location = storage.prepare_location(untrusted_location)

PARAMETER	DESCRIPTION
location	initial location provided by user TYPE: str
sample	optional Upload object that can be used by transformers. TYPE: Upload | None DEFAULT: None
**kwargs	additional parameters for transformers TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Location	transformed location

RAISES	DESCRIPTION
LocationTransformerError	when transformer is not found

file_as_upload(data, **kwargs)

Make an Upload with file content.

PARAMETER	DESCRIPTION
data	The FileData object to wrap into Upload TYPE: FileData
**kwargs	Additional metadata for the upload. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Upload	Upload object with file content

upload(location, upload, /, **kwargs)

Upload file using single stream.

Requires CREATE capability.

This is the simplest way to upload file into the storage. It uses single stream to transfer the whole file. If upload fails, no file is created in the storage.

Content is not modified during upload, it is written as-is. And content can be of size 0, which results in empty file in the storage.

When file already exists, behavior depends on override_existing setting. If it is False, ExistingFileError is raised. If it is True, existing file is replaced with new content. In this case, it is possible to lose existing file if upload fails. When adapter does not support removal, but supports overrides, this can be used to wipe the content of the file.

Location can contain nested path as long as it does not go outside of the path from settings. For example, if path is set to /data, location can be file.txt or 2024/file.txt but not ../file.txt or /etc/passwd. Violating this rule leads to LocationError.

PARAMETER	DESCRIPTION
location	The destination location for the upload. TYPE: Location
upload	The Upload object containing the file data. TYPE: Upload
**kwargs	Additional metadata for the upload. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	FileData object with details about the uploaded file.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support CREATE operation
ExistingFileError	when file already exists and override_existing is False
LocationError	when location is outside of the storage's path

resumable_start(location, size, /, **kwargs)

Prepare everything for resumable upload.

Warning

This operation is not stabilized yet.

Requires RESUMABLE capability.

content_type and hash are optional. When any of those provided, it will be used to verify the integrity of the upload. If they are missing, the upload will be accepted without verification.

PARAMETER	DESCRIPTION
location	The destination location for the upload. TYPE: Location
size	The total size of the upload in bytes. TYPE: int
**kwargs	Additional metadata for the upload. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	details about the upload.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support RESUMABLE operation

resumable_refresh(data, /, **kwargs)

Show details of the incomplete resumable upload.

Warning

This operation is not stabilized yet.

Requires RESUMABLE capability.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
**kwargs	Additional metadata for the upload. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	details about the upload.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support RESUMABLE operation

resumable_resume(data, upload, /, **kwargs)

Resume the interrupted resumable upload.

Warning

This operation is not stabilized yet.

Requires RESUMABLE capability.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
upload	The Upload object containing the content. TYPE: Upload
**kwargs	Additional metadata for the upload. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	details about the upload.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support RESUMABLE operation

resumable_remove(data, /, **kwargs)

Remove incomplete resumable upload.

Warning

This operation is not stabilized yet.

Requires RESUMABLE capability.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
**kwargs	Additional metadata for the upload. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
bool	True if upload was removed, False otherwise.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support RESUMABLE operation

multipart_start(location, size, /, **kwargs)

Prepare everything for multipart upload.

Warning

This operation is not stabilized yet.

Requires MULTIPART capability.

content_type and hash are optional. When any of those provided, it will be used to verify the integrity of the upload. If they are missing, the upload will be accepted without verification.

PARAMETER	DESCRIPTION
location	The destination location for the upload. TYPE: Location
size	The total size of the upload in bytes. TYPE: int
**kwargs	Additional metadata for the upload. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	details about the upload.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support MULTIPART operation

multipart_refresh(data, /, **kwargs)

Show details of the incomplete upload.

Warning

This operation is not stabilized yet.

Requires MULTIPART capability.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
**kwargs	Additional metadata for the upload. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	details about the upload.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support MULTIPART operation

multipart_update(data, upload, part, /, **kwargs)

Add data to the incomplete upload.

Warning

This operation is not stabilized yet.

Requires MULTIPART capability.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
upload	The Upload object containing the content. TYPE: Upload
part	Position of the given part among other parts, starting with 0. TYPE: int
**kwargs	Additional metadata for the upload. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	details about the upload.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support MULTIPART operation

multipart_complete(data, /, **kwargs)

Verify file integrity and finalize incomplete upload.

Warning

This operation is not stabilized yet.

Requires MULTIPART capability.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
**kwargs	Additional metadata for the upload. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	details about the upload.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support MULTIPART operation

multipart_remove(data, /, **kwargs)

Interrupt and remove incomplete upload.

Warning

This operation is not stabilized yet.

Requires MULTIPART capability.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
**kwargs	Additional metadata for the upload. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
bool	True if upload was removed, False otherwise.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support MULTIPART operation

exists(data, /, **kwargs)

Check if file exists in the storage.

Requires EXISTS capability.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to check. TYPE: FileData
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
bool	True if file exists, False otherwise.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support EXISTS operation

remove(data, /, **kwargs)

Remove file from the storage.

Requires REMOVE capability.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to remove. TYPE: FileData
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
bool	True if file was removed, False otherwise.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support REMOVE operation

scan(**kwargs)

List all locations(filenames) in storage.

Requires SCAN capability.

This operation lists all locations (filenames) in the storage if they start with the configured path. If path is empty, all locations are listed. Locations that match path partially(e.g. location nested_dir overlaps with path nested) are not listed.

PARAMETER	DESCRIPTION
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Iterable[str]	An iterable of location strings.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support SCAN operation

analyze(location, /, **kwargs)

Return file details.

Requires ANALYZE capability.

FileData produced by this operation is the same as data produced by the upload().

Attempt to analyze non-existing location leads to MissingFileError

PARAMETER	DESCRIPTION
location	The location of the file to analyze. TYPE: Location
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	details about the file.

RAISES	DESCRIPTION
MissingFileError	when location does not exist

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support ANALYZE operation

signed(action, duration, location, **kwargs)

Make an URL for signed action.

Warning

This operation is not stabilized yet.

Requires SIGNED capability.

This operation creates a signed URL that allows performing the specified action (e.g., "upload", "download") on the given location for a limited duration. The signed URL is typically used to grant temporary access to a file without requiring authentication. The URL includes a signature that verifies its authenticity and validity.

Depending on the action, user is expected to use the URL in different ways:

• upload - use HTTP PUT request to send the file content to the URL.
• download - use HTTP GET request to retrieve the file content from the URL.
• delete - use HTTP DELETE request to remove the file at the URL.

Check adapter specific implementation for details. Some actions may not be supported or require additional information. For example, upload with Azure Blob Storage requires header x-ms-blob-type: BlockBlob.

PARAMETER	DESCRIPTION
action	The action to sign (upload, download, delete). TYPE: SignedAction
duration	The duration for which the signed URL is valid. TYPE: int
location	The location of the file to sign. TYPE: Location
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
str	A signed URL as a string.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support SIGNED operation

stream(data, /, **kwargs)

Return byte-stream of the file content.

Requires STREAM capability.

Returns iterable that yields chunks of bytes from the file. The size of each chunk depends on the storage implementation. The iterable can be used in a for loop or converted to a list to get all chunks at once.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to stream. TYPE: FileData
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Iterable[bytes]	An iterable yielding chunks of bytes from the file.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support STREAM operation
MissingFileError	when file does not exist

range(data, start=0, end=None, /, **kwargs)

Return slice of the file content.

Requires RANGE capability.

This operation slices the file content from the specified start byte offset to the end byte offset (exclusive). If end is None, the slice extends to the end of the file. The operation returns an iterable that yields chunks of bytes from the specified range. The size of each chunk depends on the storage implementation. The iterable can be used in a for loop or converted to a list to get all chunks at once.

Unlike python slices, this operation does not expect negative indexes, but certain adapters may support it.

Attempt to get a range from non-existing file leads to MissingFileError.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to read. TYPE: FileData
start	The starting byte offset. TYPE: int DEFAULT: 0
end	The ending byte offset (inclusive). TYPE: int | None DEFAULT: None
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support RANGE operation
MissingFileError	when file does not exist

range_synthetic(data, start=0, end=None, /, **kwargs)

Generic implementation of range operation that relies on STREAM.

This method provides a generic implementation of the range operation using the STREAM capability. It reads the file in chunks and yields only the specified range of bytes.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to read. TYPE: FileData
start	The starting byte offset. TYPE: int DEFAULT: 0
end	The ending byte offset (inclusive). TYPE: int | None DEFAULT: None
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support STREAM operation
MissingFileError	when file does not exist

content(data, /, **kwargs)

Return file content as a single byte object.

Requires STREAM capability.

Returns complete file content as a single bytes object. This method internally uses stream() method to read the file content in chunks and then combines those chunks into a single bytes object.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to read. TYPE: FileData
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
bytes	The complete file content as a single bytes object.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support STREAM operation
MissingFileError	when file does not exist

append(data, upload, /, **kwargs)

Append content to existing file.

Requires APPEND capability.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to append to. TYPE: FileData
upload	The Upload object containing the content to append. TYPE: Upload
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	Updated FileData object with details about the file after appending.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support APPEND operation
MissingFileError	when file does not exist

copy(location, data, /, **kwargs)

Copy file inside the storage.

Requires COPY capability.

This operation creates a duplicate of the specified file at a new location within the same storage. The copied file retains the same content and metadata as the original file.

If file already exists at the destination location, behavior depends on override_existing setting. If it is False, ExistingFileError is raised. If it is True, existing file is replaced with the copied file. In this case, it is possible to lose existing file if copy fails.

PARAMETER	DESCRIPTION
location	The destination location for the copied file. TYPE: Location
data	The FileData object representing the file to copy. TYPE: FileData
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	FileData object with details about the copied file.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support COPY operation
MissingFileError	when source file does not exist
ExistingFileError	when destination file already exists and override_existing is False

copy_synthetic(location, data, dest_storage, /, **kwargs)

Generic implementation of the copy operation that relies on CREATE.

This method provides a generic implementation of the copy operation using the CREATE capability of the destination storage and the STREAM capability of the source storage. It reads the file content from the source storage and uploads it to the destination storage.

PARAMETER	DESCRIPTION
location	The destination location for the copied file. TYPE: Location
data	The FileData object representing the file to copy. TYPE: FileData
dest_storage	The destination storage TYPE: Storage
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support CREATE operation
MissingFileError	when source file does not exist
ExistingFileError	when destination file already exists and override_existing is False

move(location, data, /, **kwargs)

Move file to a different location inside the storage.

Requires MOVE capability.

This operation relocates the specified file to a new location within the same storage. After the move operation, the file will no longer exist at its original location.

PARAMETER	DESCRIPTION
location	The destination location for the moved file. TYPE: Location
data	The FileData object representing the file to move. TYPE: FileData
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	FileData object with details about the moved file.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support MOVE operation
MissingFileError	when source file does not exist
ExistingFileError	when destination file already exists and override_existing is False

move_synthetic(location, data, dest_storage, /, **kwargs)

Generic implementation of move operation.

Relies on CREATE and REMOVE.

PARAMETER	DESCRIPTION
location	The destination location for the moved file. TYPE: Location
data	The FileData object representing the file to move. TYPE: FileData
dest_storage	The destination storage TYPE: Storage
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	FileData object with details about the moved file.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support CREATE or REMOVE operation
MissingFileError	when source file does not exist
ExistingFileError	when destination file already exists and override_existing is False

compose(location, /, *files, **kwargs)

Combine multiple files into a new file.

Requires COMPOSE capability.

This operation combines multiple source files into a single destination file. The content of the source files is concatenated in the order they are provided to form the content of the new file.

PARAMETER	DESCRIPTION
location	The destination location for the composed file. TYPE: Location
*files	FileData objects representing the files to combine. TYPE: FileData DEFAULT: ()
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
FileData	FileData object with details about the composed file.

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support COMPOSE operation
MissingFileError	when any of source files do not exist
ExistingFileError	when destination file already exists and override_existing is False

compose_synthetic(location, dest_storage, /, *files, **kwargs)

Generic composition that relies on APPEND.

This method provides a generic implementation of the compose operation using the APPEND capability of the destination storage. It creates an empty file at the destination location and then appends the content of each source file to it in sequence.

PARAMETER	DESCRIPTION
location	The destination location for the composed file. TYPE: Location
dest_storage	The destination storage TYPE: Storage
*files	FileData objects representing the files to combine. TYPE: FileData DEFAULT: ()
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RAISES	DESCRIPTION
UnsupportedOperationError	when storage does not support APPEND operation
MissingFileError	when any of source files do not exist
ExistingFileError	when destination file already exists and override_existing is False

one_time_link(data, /, **kwargs)

Return one-time download link.

Requires LINK_ONE_TIME capability.

PARAMETER	DESCRIPTION
data	The FileData object representing the file. TYPE: FileData
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
str | None	A one-time download link as a string, or None if not supported.

temporal_link(data, duration, /, **kwargs)

Return temporal download link.

Requires LINK_TEMPORAL capability.

PARAMETER	DESCRIPTION
data	The FileData object representing the file. TYPE: FileData
duration	The duration for which the link is valid. TYPE: int
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
str | None	A temporal download link as a string, or None if not supported.

permanent_link(data, /, **kwargs)

Return permanent download link.

Requires LINK_PERMANENT capability.

PARAMETER	DESCRIPTION
data	The FileData object representing the file. TYPE: FileData
**kwargs	Additional metadata for the operation. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
str | None	A permanent download link as a string, or None if not supported.

Capability

Bases: Flag

Enumeration of operations supported by the storage.

Example

read_and_write = Capability.STREAM | Capability.CREATE
if storage.supports(read_and_write)
    ...

ANALYZE = enum.auto() class-attribute instance-attribute

Return file details from the storage.

APPEND = enum.auto() class-attribute instance-attribute

Add content to the existing file.

COMPOSE = enum.auto() class-attribute instance-attribute

Combine multiple files into a new one in the same storage.

COPY = enum.auto() class-attribute instance-attribute

Make a copy of the file inside the same storage.

CREATE = enum.auto() class-attribute instance-attribute

Create a file as an atomic object.

EXISTS = enum.auto() class-attribute instance-attribute

Check if file exists.

LINK_ONE_TIME = enum.auto() class-attribute instance-attribute

Make one-time download link.

LINK_PERMANENT = enum.auto() class-attribute instance-attribute

Make permanent download link.

LINK_TEMPORAL = enum.auto() class-attribute instance-attribute

Make expiring download link.

MOVE = enum.auto() class-attribute instance-attribute

Move file to a different location inside the same storage.

MULTIPART = enum.auto() class-attribute instance-attribute

Create file in 3 stages: initialize, upload(repeatable), complete.

RANGE = enum.auto() class-attribute instance-attribute

Return specific range of bytes from the file.

REMOVE = enum.auto() class-attribute instance-attribute

Remove file from the storage.

RESUMABLE = enum.auto() class-attribute instance-attribute

Perform resumable uploads that can be continued after interruption.

SCAN = enum.auto() class-attribute instance-attribute

Iterate over all files in the storage.

SIGNED = enum.auto() class-attribute instance-attribute

Generate signed URL for specific operation.

STREAM = enum.auto() class-attribute instance-attribute

Return file content as stream of bytes.

can(operation)

Check whether the cluster supports given operation.

exclude(*capabilities)

Remove capabilities from the cluster.

PARAMETER	DESCRIPTION
capabilities	removed capabilities TYPE: Capability

Example

cluster = cluster.exclude(Capability.REMOVE)

Settings dataclass

Settings for the storage adapter.

disabled_capabilities = cast('list[str]', dataclasses.field(default_factory=list)) class-attribute instance-attribute

Capabilities that are not supported even if implemented.

initialize = False class-attribute instance-attribute

Prepare storage backend for uploads(create path, bucket, DB)

location_transformers = cast('list[str]', dataclasses.field(default_factory=list)) class-attribute instance-attribute

List of transformations applied to the file location.

name = 'unknown' class-attribute instance-attribute

Descriptive name of the storage used for debugging.

override_existing = False class-attribute instance-attribute

If file already exists, replace it with new content.

path = '' class-attribute instance-attribute

Prefix for the file's location.

from_dict(data) classmethod

Make settings object using dictionary as a source.

Any unexpected options are extracted from the data to avoid initialization errors from dataclass constructor.

PARAMETER	DESCRIPTION
data	mapping with settings TYPE: Mapping[str, Any]

RETURNS	DESCRIPTION
Settings	settings object built from data

Uploader

Bases: StorageService

Service responsible for writing data into a storage.

Storage internally calls methods of this service. For example, Storage.upload(location, upload, **kwargs) results in Uploader.upload(location, upload, kwargs).

Example

class MyUploader(Uploader):
    capabilities = Capability.CREATE

    def upload(
        self, location: types.Location, upload: Upload, extras: dict[str, Any]
    ) -> FileData:
        reader = upload.hashing_reader()

        with open(location, "wb") as dest:
            dest.write(reader.read())

        return FileData(
            location, upload.size,
            upload.content_type,
            reader.get_hash()
        )

multipart_complete(data, extras)

Verify file integrity and finalize incomplete upload.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
extras	Additional metadata for the upload. TYPE: dict[str, Any]

multipart_refresh(data, extras)

Show details of the incomplete upload.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
extras	Additional metadata for the upload. TYPE: dict[str, Any]

multipart_remove(data, extras)

Interrupt and remove incomplete upload.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
extras	Additional metadata for the upload. TYPE: dict[str, Any]

multipart_start(location, size, extras)

Prepare everything for multipart(resumable) upload.

PARAMETER	DESCRIPTION
location	The destination location for the upload. TYPE: Location
size	Expected upload size. TYPE: int
extras	Additional metadata for the upload. TYPE: dict[str, Any]

multipart_update(data, upload, part, extras)

Add data to the incomplete upload.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
upload	The Upload object containing the content. TYPE: Upload
part	Position of the given part among other parts, starting with 0. TYPE: int
extras	Additional metadata for the upload. TYPE: dict[str, Any]

resumable_refresh(data, extras)

Show details of the incomplete resumable upload.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
extras	Additional metadata for the upload. TYPE: dict[str, Any]

resumable_remove(data, extras)

Remove incomplete resumable upload.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
extras	Additional metadata for the upload. TYPE: dict[str, Any]

resumable_resume(data, upload, extras)

Resume the interrupted resumable upload.

PARAMETER	DESCRIPTION
data	The FileData object containing the upload metadata. TYPE: FileData
upload	The Upload object containing the content. TYPE: Upload
extras	Additional metadata for the upload. TYPE: dict[str, Any]

resumable_start(location, size, extras)

Prepare everything for resumable upload.

PARAMETER	DESCRIPTION
location	The destination location for the upload. TYPE: Location
size	Expected upload size. TYPE: int
extras	Additional metadata for the upload. TYPE: dict[str, Any]

upload(location, upload, extras)

Upload file using single stream.

PARAMETER	DESCRIPTION
location	The destination location for the upload. TYPE: Location
upload	The Upload object containing the file data. TYPE: Upload
extras	Additional metadata for the upload. TYPE: dict[str, Any]

Manager

Bases: StorageService

Service responsible for maintenance file operations.

Storage internally calls methods of this service. For example, Storage.remove(data, **kwargs) results in Manager.remove(data, kwargs).

Example

class MyManager(Manager):
    capabilities = Capability.REMOVE
    def remove(
        self, data: FileData|FileData, extras: dict[str, Any]
    ) -> bool:
        os.remove(data.location)
        return True

analyze(location, extras)

Return details about location.

PARAMETER	DESCRIPTION
location	The location of the file to analyze. TYPE: Location
extras	Additional metadata for the operation. TYPE: dict[str, Any]

append(data, upload, extras)

Append content to existing file.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to append to. TYPE: FileData
upload	The Upload object containing the content to append. TYPE: Upload
extras	Additional metadata for the operation. TYPE: dict[str, Any]

compose(location, datas, extras)

Combine multipe file inside the storage into a new one.

PARAMETER	DESCRIPTION
location	The destination location for the composed file. TYPE: Location
datas	An iterable of FileData objects representing the files to combine. TYPE: Iterable[FileData]
extras	Additional metadata for the operation. TYPE: dict[str, Any]

copy(location, data, extras)

Copy file inside the storage.

PARAMETER	DESCRIPTION
location	The destination location for the copied file. TYPE: Location
data	The FileData object representing the file to copy. TYPE: FileData
extras	Additional metadata for the operation. TYPE: dict[str, Any]

exists(data, extras)

Check if file exists in the storage.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to check. TYPE: FileData
extras	Additional metadata for the operation. TYPE: dict[str, Any]

move(location, data, extras)

Move file to a different location inside the storage.

PARAMETER	DESCRIPTION
location	The destination location for the moved file. TYPE: Location
data	The FileData object representing the file to move. TYPE: FileData
extras	Additional metadata for the operation. TYPE: dict[str, Any]

remove(data, extras)

Remove file from the storage.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to remove. TYPE: FileData
extras	Additional metadata for the operation. TYPE: dict[str, Any]

scan(extras)

List all locations(filenames) in storage.

PARAMETER	DESCRIPTION
extras	Additional metadata for the operation. TYPE: dict[str, Any]

signed(action, duration, location, extras)

Make an URL for signed action.

PARAMETER	DESCRIPTION
action	The action to sign (e.g., "upload", "download"). TYPE: SignedAction
duration	The duration for which the signed URL is valid. TYPE: int
location	The location of the file to sign. TYPE: Location
extras	Additional metadata for the operation. TYPE: dict[str, Any]

Reader

Bases: StorageService

Service responsible for reading data from the storage.

Storage internally calls methods of this service. For example, Storage.stream(data, **kwargs) results in Reader.stream(data, kwargs).

Example

class MyReader(Reader):
    capabilities = Capability.STREAM

    def stream(
        self, data: data.FileData, extras: dict[str, Any]
    ) -> Iterable[bytes]:
        return open(data.location, "rb")

content(data, extras)

Return file content as a single byte object.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to read. TYPE: FileData
extras	Additional metadata for the operation. TYPE: dict[str, Any]

one_time_link(data, extras)

Return one-time download link.

PARAMETER	DESCRIPTION
data	The FileData object representing the file. TYPE: FileData
extras	Additional metadata for the operation. TYPE: dict[str, Any]

permanent_link(data, extras)

Return permanent download link.

PARAMETER	DESCRIPTION
data	The FileData object representing the file. TYPE: FileData
extras	Additional metadata for the operation. TYPE: dict[str, Any]

range(data, start, end, extras)

Return slice of the file content.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to read. TYPE: FileData
start	The starting byte offset. TYPE: int
end	The ending byte offset (inclusive). TYPE: int | None
extras	Additional metadata for the operation. TYPE: dict[str, Any]

stream(data, extras)

Return byte-stream of the file content.

PARAMETER	DESCRIPTION
data	The FileData object representing the file to stream. TYPE: FileData
extras	Additional metadata for the operation. TYPE: dict[str, Any]

temporal_link(data, duration, extras)

Return temporal download link.

PARAMETER	DESCRIPTION
data	The FileData object representing the file. TYPE: FileData
duration	The duration for which the link is valid. TYPE: int
extras	Additional metadata for the operation. TYPE: dict[str, Any]

FileData dataclass

Bases: BaseData

Information required by storage to operate the file.

PARAMETER	DESCRIPTION
location	filepath, filename or any other type of unique identifier TYPE: Location
size	size of the file in bytes TYPE: int DEFAULT: 0
content_type	MIMEtype of the file TYPE: str DEFAULT: 'application/octet-stream'
hash	checksum of the file TYPE: str DEFAULT: ''
storage_data	additional details set by storage adapter TYPE: dict[str, Any] DEFAULT: cast('dict[str, Any]', field(default_factory=dict))

Example

FileData(
    "local/path.txt",
    123,
    "text/plain",
    md5_of_content,
)

Upload dataclass

Standard upload details.

PARAMETER	DESCRIPTION
stream	iterable of bytes or file-like object TYPE: PStream
filename	name of the file TYPE: str
size	size of the file in bytes TYPE: int
content_type	MIMEtype of the file TYPE: str

Example

>>> upload = Upload(
...     BytesIO(b"hello world"),
...     "file.txt",
...     11,
...     "text/plain",
... )
>>>
>>> upload.size
11
>>> upload.filename
'file.txt'
>>> upload.content_type
'text/plain'
>>> upload.stream.read()
b'hello world'

content_type instance-attribute

MIME Type of the file

filename instance-attribute

Name of the file

seekable_stream property

Return stream that can be rewinded after reading.

If internal stream does not support file-like seek, nothing is returned from this property.

Use this property if you want to read the file ahead, to get CSV column names, list of files inside ZIP, EXIF metadata. If you get None from it, stream does not support seeking and you won't be able to rewind cursor to the beginning of the file after reading something.

Example

upload = make_upload(...)
if fd := upload.seekable_stream():
    # read fragment of the file
    chunk = fd.read(1024)
    # move cursor to the end of the stream
    fd.seek(0, 2)
    # position of the cursor is the same as number of bytes in stream
    size = fd.tell()
    # move cursor back, because you don't want to accidentally loose
    # any bites from the beginning of stream when uploader reads from it
    fd.seek(0)

RETURNS	DESCRIPTION
PSeekableStream | None	file-like stream or nothing

size instance-attribute

Size of the file

stream instance-attribute

Content as iterable of bytes

hashing_reader(**kwargs)

Get reader for the upload that computes hash while reading content.

Location = NewType('Location', str) module-attribute

SignedAction = Literal['upload', 'download', 'delete'] module-attribute

HashingReader

IO stream wrapper that computes content hash while stream is consumed.

PARAMETER	DESCRIPTION
stream	iterable of bytes or file-like object TYPE: PStream
chunk_size	max number of bytes read at once TYPE: int DEFAULT: CHUNK_SIZE
algorithm	hashing algorithm TYPE: str DEFAULT: 'md5'

Example

reader = HashingReader(readable_stream)
for chunk in reader:
    ...
print(f"Hash: {reader.get_hash()}")

exhaust()

Exhaust internal stream to compute final version of content hash.

Note, this method does not returns data from the stream. The content will be irreversibly lost after method execution.

get_hash()

Get current content hash as a string.

read()

Read and return all bytes from stream at once.

Registry

Bases: Generic[V, K]

Mutable collection of objects.

Example

col = Registry()

col.register("one", 1)
assert col.get("one") == 1

col.reset()
assert col.get("one") is None

assert list(col) == ["one"]

collect()

Collect members of the registry.

decorated(key)

Collect member via decorator.

get(key)

Get the optional member from registry.

pop(key)

Remove the member from registry.

register(key, member)

Add a member to registry.

reset()

Remove all members from registry.

IterableBytesReader

Bases: AbstractReader[Iterable[int]]

Wrapper that transforms iterable of bytes into readable stream.

Example

The simplest iterable of bytes is a list that contains byte strings:

parts = [b"hello", b" ", b"world"]
reader = IterableBytesReader(parts)
assert reader.read() == b"hello world"

More realistic scenario is wrapping generator that produces byte string in order to initialize Upload:

def data_generator():
    yield b"hello"
    yield b" "
    yield b"world"

stream = IterableBytesReader(data_generator())
upload = Upload(stream, "my_file.txt", 11, "text/plain")

adapters = Registry['type[Storage]']() module-attribute

parse_filesize(value)

Transform human-readable filesize into an integer.

PARAMETER	DESCRIPTION
value	human-readable filesize TYPE: str

RAISES	DESCRIPTION
ValueError	size cannot be parsed or uses unknown units

Example

size = parse_filesize("10GiB")
assert size == 10_737_418_240

humanize_filesize(value, base=SI_BASE)

Transform an integer into human-readable filesize.

PARAMETER	DESCRIPTION
value	size in bytes TYPE: int | float
base	1000 for SI(KB, MB) or 1024 for binary(KiB, MiB) TYPE: int DEFAULT: SI_BASE

RAISES	DESCRIPTION
ValueError	base is not recognized

Example

size = humanize_filesize(10_737_418_240, base=1024)
assert size == "10GiB"
size = humanize_filesize(10_418_240, base=1024)
assert size == "9.93MiB"