`tlc.core.url_adapter`#

Provides a unified interface for URL-based file operations across different schemes and storage backends.

Module Contents#

Classes#

Class	Description
`IfExistsOption`	An Enum indicating what to do if content already exists at the target URL
`UrlAdapterDirEntry`	The base class for all directory entries.
`UrlAdapter`	The base class for all URL adapters
`UrlAdapterAsyncFromSync`	A UrlAdapter where the Async methods are implemented as Futures that call the equivalent Sync methods on the thread pool
`UrlAdapterSyncFromAsync`	A UrlAdapter where the Sync methods are implemented by calling the equivalent Async methods and then waiting on the Futures returned.

API#

class tlc.core.url_adapter.IfExistsOption#

Bases: str, enum.Enum

An Enum indicating what to do if content already exists at the target URL

from tlc.core.url_adapter import IfExistsOption, UrlAdapter
from tlc.core.url import Url

url = Url("s3://my-bucket/my-file.txt")
adapter = UrlAdapter.get_adapter(url)
adapter.write_string_content_to_url(url, "Hello World!", IfExistsOption.OVERWRITE)

Initialize self. See help(type(self)) for accurate signature.

OVERWRITE = overwrite#: If content already exists at the target URL, overwrite the content (default).

RENAME = rename#: If content already exists at the target URL, leave that content, and write the new content to a new URL that is similar to the original target URL.

RAISE = raise#: If content already exists at the target URL, raise an exception.

REUSE = reuse#: If content already exists at the target URL, reuse the existing content.

class tlc.core.url_adapter.UrlAdapterDirEntry#

The base class for all directory entries.

abstract property name: str#: The entry’s base filename

abstract property path: str#: The entry’s full path name

abstract is_dir() → bool#: Return True if this entry is a directory

abstract is_file() → bool#: Return True if this entry is a file

abstract mtime() → Any#

Return the modification time object for this entry

The mtime object is not specified to be of any particular type, but must be comparable to other mtime objects of the same type

May raise exceptions when accessing the resources

abstract mtime_datetime() → datetime.datetime#: Return the modification time for this entry as a datetime object

class tlc.core.url_adapter.UrlAdapter#

Bases: abc.ABC

The base class for all URL adapters

abstract schemes() → list[tlc.core.url.Scheme]#: Get URL schemes

abstract read_string_content_from_url(url: tlc.core.url.Url) → str#: Read content from URL synchronously, dispatch

abstract read_string_content_from_url_async(url: tlc.core.url.Url) → concurrent.futures.Future#: Read content from URL asynchronously

abstract read_binary_content_from_url(url: tlc.core.url.Url) → bytes#: Read binary content from URL synchronously, dispatch

abstract read_binary_content_from_url_async(url: tlc.core.url.Url) → concurrent.futures.Future#: Read binary content from URL asynchronously

write_string_content_to_url(url: tlc.core.url.Url, content: str, options: tlc.core.url_adapter.IfExistsOption = IfExistsOption.OVERWRITE) → tlc.core.url.Url#: Write content to URL synchronously

abstract write_string_content_to_url_async(url: tlc.core.url.Url, content: str, options: tlc.core.url_adapter.IfExistsOption = IfExistsOption.OVERWRITE) → concurrent.futures.Future#: Write content to URL asynchronously

write_binary_content_to_url(url: tlc.core.url.Url, content: bytes, options: tlc.core.url_adapter.IfExistsOption = IfExistsOption.OVERWRITE) → tlc.core.url.Url#

Write binary content to URL synchronously

Handles write options and dispatches to _write_binary_content_to_url

abstract write_binary_content_to_url_async(url: tlc.core.url.Url, content: bytes, options: tlc.core.url_adapter.IfExistsOption = IfExistsOption.OVERWRITE) → concurrent.futures.Future#: Write binary content to URL asynchronously

copy_url(source: tlc.core.url.Url, destination: tlc.core.url.Url, options: tlc.core.url_adapter.IfExistsOption = IfExistsOption.OVERWRITE) → tlc.core.url.Url#: Copy URL synchronously

abstract copy_url_async(source: tlc.core.url.Url, destination: tlc.core.url.Url, options: tlc.core.url_adapter.IfExistsOption = IfExistsOption.OVERWRITE) → concurrent.futures.Future#: Copy URL asynchronously

abstract delete_url(url: tlc.core.url.Url) → None#: Delete URL synchronously

abstract delete_url_async(url: tlc.core.url.Url) → concurrent.futures.Future#: Delete URL asynchronously

abstract make_dirs(url: tlc.core.url.Url, exist_ok: bool = False) → None#

Create a leaf directory and all intermediate ones synchronously

For flat file hierarchies, this may be a no-op, see is_file_hierarchy_flat.

abstract make_dirs_async(url: tlc.core.url.Url, exist_ok: bool = False) → concurrent.futures.Future#

Create a leaf directory and all intermediate ones asynchronously

For flat file hierarchies, this may be a no-op, see is_file_hierarchy_flat.

abstract get_file_size(url: tlc.core.url.Url) → int#: Get the size of the file at the given URL synchronously

abstract get_file_size_async(url: tlc.core.url.Url) → concurrent.futures.Future#: Get the size of the file at the given URL asynchronously

abstract list_dir(url: tlc.core.url.Url) → Iterator[tlc.core.url_adapter.UrlAdapterDirEntry]#: List the entries belonging to the directory at the given URL synchronously

abstract list_dir_async(url: tlc.core.url.Url) → concurrent.futures.Future#: List the entries belonging to the directory at the given URL asynchronously

abstract exists(url: tlc.core.url.Url) → bool#: Return True if the given URL refers to an existing path synchronously

abstract exists_async(url: tlc.core.url.Url) → concurrent.futures.Future#: Return True if the given URL refers to an existing path asynchronously

abstract is_dir(url: tlc.core.url.Url) → bool#: Return True if the given URL refers to a directory synchronously

abstract is_dir_async(url: tlc.core.url.Url) → concurrent.futures.Future#: Return True if the given URL refers to a directory asynchronously

abstract is_writable(url: tlc.core.url.Url) → bool#: Return True if the given URL is writable

abstract is_writable_async(url: tlc.core.url.Url) → concurrent.futures.Future#: Return True if the given URL is writable asynchronously

is_file_hierarchy_flat() → bool#

Determine if the file hierarchy is flat for this adapter

Cloud storage normally has a flat hierarchy, meaning that make_dirs is a no-op.

class tlc.core.url_adapter.UrlAdapterAsyncFromSync#

Bases: tlc.core.url_adapter.UrlAdapter

A UrlAdapter where the Async methods are implemented as Futures that call the equivalent Sync methods on the thread pool

Derived classes must implement the Sync methods.

read_string_content_from_url_async(url: tlc.core.url.Url) → concurrent.futures.Future#

read_binary_content_from_url_async(url: tlc.core.url.Url) → concurrent.futures.Future#

write_string_content_to_url_async(url: tlc.core.url.Url, content: str, options: tlc.core.url_adapter.IfExistsOption = IfExistsOption.OVERWRITE) → concurrent.futures.Future#

write_binary_content_to_url_async(url: tlc.core.url.Url, content: bytes, options: tlc.core.url_adapter.IfExistsOption = IfExistsOption.OVERWRITE) → concurrent.futures.Future#

delete_url_async(url: tlc.core.url.Url) → concurrent.futures.Future#

copy_url_async(source: tlc.core.url.Url, destination: tlc.core.url.Url, options: tlc.core.url_adapter.IfExistsOption = IfExistsOption.OVERWRITE) → concurrent.futures.Future#

make_dirs_async(url: tlc.core.url.Url, exist_ok: bool = False) → concurrent.futures.Future#

get_file_size_async(url: tlc.core.url.Url) → concurrent.futures.Future#

list_dir_async(url: tlc.core.url.Url) → concurrent.futures.Future#

exists_async(url: tlc.core.url.Url) → concurrent.futures.Future#

is_dir_async(url: tlc.core.url.Url) → concurrent.futures.Future#

is_writable_async(url: tlc.core.url.Url) → concurrent.futures.Future#

class tlc.core.url_adapter.UrlAdapterSyncFromAsync#

Bases: tlc.core.url_adapter.UrlAdapter

A UrlAdapter where the Sync methods are implemented by calling the equivalent Async methods and then waiting on the Futures returned.

Derived classes must implement the Async methods.

read_string_content_from_url(url: tlc.core.url.Url) → str#

read_binary_content_from_url(url: tlc.core.url.Url) → bytes#

write_string_content_to_url(url: tlc.core.url.Url, content: str, options: tlc.core.url_adapter.IfExistsOption = IfExistsOption.OVERWRITE) → tlc.core.url.Url#

write_binary_content_to_url(url: tlc.core.url.Url, content: bytes, options: tlc.core.url_adapter.IfExistsOption = IfExistsOption.OVERWRITE) → tlc.core.url.Url#

delete_url(url: tlc.core.url.Url) → None#

make_dirs(url: tlc.core.url.Url, exist_ok: bool = False) → None#

get_file_size(url: tlc.core.url.Url) → int#

list_dir(url: tlc.core.url.Url) → Iterator[tlc.core.url_adapter.UrlAdapterDirEntry]#

exists(url: tlc.core.url.Url) → bool#

tlc.core.url_adapter#

Module Contents#

Classes#

API#

`tlc.core.url_adapter`#