tlcurl.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

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

UrlAdapterDirEntry

The base class for all directory entries.

UrlAdapterSyncFromAsync

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

Data

Data

Description

logger

API

class 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).

RAISE = raise

If content already exists at the target URL, raise an exception.

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.

REUSE = reuse

If content already exists at the target URL, reuse the existing content.

class UrlAdapter

Bases: abc.ABC

The base class for all URL adapters

copy_url(
source: Url,
destination: Url,
options: IfExistsOption = IfExistsOption.OVERWRITE,
) → Url

Copy URL synchronously

abstract copy_url_async(
source: Url,
destination: Url,
options: IfExistsOption = IfExistsOption.OVERWRITE,
) → Future

Copy URL asynchronously

abstract delete_url(
url: Url,
) → None

Delete URL synchronously

abstract delete_url_async(
url: Url,
) → Future

Delete URL asynchronously

abstract exists(
url: Url,
) → bool

Return True if the given URL refers to an existing path synchronously

abstract exists_async(
url: Url,
) → Future

Return True if the given URL refers to an existing path asynchronously

abstract get_file_size(
url: Url,
) → int

Get the size of the file at the given URL synchronously

abstract get_file_size_async(
url: Url,
) → Future

Get the size of the file at the given URL asynchronously

abstract is_dir(
url: Url,
) → bool

Return True if the given URL refers to a directory synchronously

abstract is_dir_async(
url: Url,
) → Future

Return True if the given URL refers to a directory 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.

abstract is_writable(
url: Url,
) → bool

Return True if the given URL is writable

abstract is_writable_async(
url: Url,
) → Future

Return True if the given URL is writable asynchronously

abstract list_dir(
url: Url,
) → Iterator[UrlAdapterDirEntry]

List the entries belonging to the directory at the given URL synchronously

abstract list_dir_async(
url: Url,
) → Future

List the entries belonging to the directory at the given URL asynchronously

abstract make_dirs(
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: Url,
exist_ok: bool = False,
) → 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 read_binary_content_from_url(
url: Url,
) → bytes

Read binary content from URL synchronously, dispatch

abstract read_binary_content_from_url_async(
url: Url,
) → Future

Read binary content from URL asynchronously

abstract read_string_content_from_url(
url: Url,
encoding: str = 'utf-8',
) → str

Read content from URL synchronously, dispatch

abstract read_string_content_from_url_async(
url: Url,
encoding: str = 'utf-8',
) → Future

Read content from URL asynchronously

abstract schemes() → list[Scheme]

Get URL schemes

abstract stat(
url: Url,
) → UrlAdapterDirEntry
Get metadata about a file or directory at the given URL synchronously

param url: The URL to get metadata for

return: A UrlAdapterDirEntry containing metadata about the file or directory

raises FileNotFoundError: If the URL does not exist

abstract stat_async(
url: Url,
) → Future
Get metadata about a file or directory at the given URL asynchronously

param url: The URL to get metadata for

return: A Future that resolves to a UrlAdapterDirEntry containing metadata

raises FileNotFoundError: If the URL does not exist

abstract touch(
url: Url,
) → None

Update the last modified timestamp of a file to the current time. Creates the file if it doesn’t exist.

Parameters:

url – The URL of the file to touch

write_binary_content_to_url(
url: Url,
content: bytes,
options: IfExistsOption = IfExistsOption.OVERWRITE,
) → 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: Url,
content: bytes,
options: IfExistsOption = IfExistsOption.OVERWRITE,
) → Future

Write binary content to URL asynchronously

write_string_content_to_url(
url: Url,
content: str,
options: IfExistsOption = IfExistsOption.OVERWRITE,
encoding: str = 'utf-8',
) → Url

Write content to URL synchronously

abstract write_string_content_to_url_async(
url: Url,
content: str,
options: IfExistsOption = IfExistsOption.OVERWRITE,
encoding: str = 'utf-8',
) → Future

Write content to URL asynchronously

class UrlAdapterAsyncFromSync

Bases: tlcurl.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.

copy_url_async(
source: Url,
destination: Url,
options: IfExistsOption = IfExistsOption.OVERWRITE,
) → Future
delete_url_async(
url: Url,
) → Future
exists_async(
url: Url,
) → Future
get_file_size_async(
url: Url,
) → Future
is_dir_async(
url: Url,
) → Future
is_writable_async(
url: Url,
) → Future
list_dir_async(
url: Url,
) → Future
make_dirs_async(
url: Url,
exist_ok: bool = False,
) → Future
read_binary_content_from_url_async(
url: Url,
) → Future
read_string_content_from_url_async(
url: Url,
encoding: str = 'utf-8',
) → Future
stat_async(
url: Url,
) → Future
write_binary_content_to_url_async(
url: Url,
content: bytes,
options: IfExistsOption = IfExistsOption.OVERWRITE,
) → Future
write_string_content_to_url_async(
url: Url,
content: str,
options: IfExistsOption = IfExistsOption.OVERWRITE,
encoding: str = 'utf-8',
) → Future
class UrlAdapterDirEntry

The base class for all directory entries.

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

Return the modification time for this entry as a datetime object

abstract property name: str

The entry’s base filename

abstract property path: str

The entry’s full path name

class UrlAdapterSyncFromAsync

Bases: tlcurl.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.

delete_url(
url: Url,
) → None
exists(
url: Url,
) → bool
get_file_size(
url: Url,
) → int
list_dir(
url: Url,
) → Iterator[UrlAdapterDirEntry]
make_dirs(
url: Url,
exist_ok: bool = False,
) → None
read_binary_content_from_url(
url: Url,
) → bytes
read_string_content_from_url(
url: Url,
encoding: str = 'utf-8',
) → str
stat(
url: Url,
) → UrlAdapterDirEntry
write_binary_content_to_url(
url: Url,
content: bytes,
options: IfExistsOption = IfExistsOption.OVERWRITE,
) → Url
write_string_content_to_url(
url: Url,
content: str,
options: IfExistsOption = IfExistsOption.OVERWRITE,
encoding: str = 'utf-8',
) → Url
logger = getLogger(...)