WebdavFileSystem#

fsspec compliant webdav file system.

class webdav4.fsspec.ReopenArgs(file: Type[WebdavFile], fs: WebdavFileSystem, path: str, blocksize: Optional[int], mode: str, size: Optional[int])[source]#

Bases: NamedTuple

Args to reopen the file.

Create new instance of ReopenArgs(file, fs, path, blocksize, mode, size)

blocksize: Optional[int]#: Alias for field number 3

file: Type[WebdavFile]#: Alias for field number 0

fs: WebdavFileSystem#: Alias for field number 1

mode: str#: Alias for field number 4

path: str#: Alias for field number 2

size: Optional[int]#: Alias for field number 5

class webdav4.fsspec.UploadFile(fs: WebdavFileSystem, path: str, mode: str = 'wb', block_size: Optional[int] = None)[source]#

Bases: SpooledTemporaryFile

UploadFile for Webdav, that uses SpooledTemporaryFile.

In Webdav, you cannot upload in chunks. Similar to http, we need some kind of protocol on top to be able to do that. Sabredav provides a way to upload chunks, but it’s not implemented in Owncloud and Nextcloud. They have a different chunking mechanism. On top of it, there are TUS implementations in newer version of Owncloud.

However, as they are only for chunking, it needs to be seen if it is possible to implement full-breadth of FileObj API through that (I doubt it)

Note that, fsspec’s put_file/pipe_file don’t use this technique and are thus faster as they don’t need buffering. It is recommended to use that if you don’t need to upload content dynamically.

Extended interface with path and fs.

close() → None[source]#: Close the file.

commit() → None[source]#

Commits the file to the given path.

As we cannot upload in chunk, this is where we really upload file.

discard() → None[source]#: Discard the file.

info() → NoReturn[source]#: Info about the file upload that is in progress.

readable() → bool[source]#: It is readable.

readinto(b: Union[bytearray, memoryview, ArrayType[Any], mmap]) → int[source]#: Read bytes into the given buffer.

readuntil(char: bytes = b'\n', blocks: Optional[int] = None) → bytes[source]#: Read until the given character is found.

seekable() → bool[source]#: It is seekable.

writable() → bool[source]#: It is writable.

class webdav4.fsspec.WebdavFile(fs: WebdavFileSystem, path: str, mode: str = 'rb', block_size: Optional[int] = None, autocommit: bool = True, cache_type: str = 'readahead', cache_options: Optional[Dict[str, str]] = None, **kwargs: Any)[source]#

Bases: AbstractBufferedFile

WebdavFile that provides file-like access to remote file.

Instantiate a file-like object with the provided options.

See fsspec for more information.

close() → None[source]#: Close stream.

isatty() → bool[source]#: Check if it is an interactive fileobj.

read(length: int = -1) → Optional[Union[str, bytes]][source]#: Read chunk of bytes.

seek(loc: int, whence: int = 0) → int[source]#: Set current file location.

class webdav4.fsspec.WebdavFileSystem(*args, **kwargs)[source]#

Bases: AbstractFileSystem

Provides access to webdav through fsspec-compliant APIs.

Instantiate WebdavFileSystem with base_url and auth.

Parameters:

base_url – base url of the server
auth – Authentication to the server Refer to HTTPX’s auth for more information.
client – Webdav client to use instead, useful for testing/mocking, or extending WebdavFileSystem.
client_opts – Extra args that are passed to Webdav Client. (refer to it’s documenting for more information).

checksum(path: str) → Optional[str][source]#: Returns checksum/etag of the path.

copy(path1: str, path2: str, recursive: bool = False, on_error: Optional[str] = None, **kwargs: Any) → None[source]#: Copy files and directories.

cp_file(path1: str, path2: str, **kwargs: Any) → None[source]#: Copy a file from one path to the other.

created(path: str) → Optional[datetime][source]#: Returns creation time/date.

info(path: str, **kwargs: Any) → Dict[str, Any][source]#: Return information about the current path.

ls(path: str, detail: bool = True, **kwargs: Any) → List[Union[str, Dict[str, Any]]][source]#: ls implementation for fsspec, see fsspec for more information.

makedirs(path: str, exist_ok: bool = False) → None[source]#: Creates directory to the given path.

mkdir(path: str, create_parents: bool = True, **kwargs: Any) → None[source]#: Create directory.

modified(path: str) → Optional[datetime][source]#: Returns last modified time/data.

mv(path1: str, path2: str, recursive: bool = False, maxdepth: Optional[bool] = None, **kwargs: Any) → None[source]#: Move a file/directory from one path to the other.

pipe_file(path: str, value: bytes, **kwargs: Any) → None[source]#: Upload the contents to given file in the remote webdav server.

put_file(lpath: PathLike[AnyStr], rpath: str, callback: Optional[Callback] = None, **kwargs: Any) → None[source]#: Copy file to remote webdav server.

put_fileobj(fobj: BinaryIO, rpath: str, callback: Optional[Callback] = None, overwrite: bool = True, size: Optional[int] = None, **kwargs: Any) → None#: Upload contents from the fileobj to the remote path.

rm(path: str, recursive: bool = False, maxdepth: Optional[int] = None) → None[source]#: Delete files and directories.

rm_file(path: str) → None[source]#: Remove a file.

rmdir(path: str) → None[source]#: Remove a directory, if empty.

sign(path: str, expiration: int = 100, **kwargs: Any) → None[source]#: Create a signed URL representing the given path.

size(path: str) → Optional[int][source]#: Returns size of the path.

upload_fileobj(fobj: BinaryIO, rpath: str, callback: Optional[Callback] = None, overwrite: bool = True, size: Optional[int] = None, **kwargs: Any) → None[source]#: Upload contents from the fileobj to the remote path.

webdav4.fsspec.reopen(args: ReopenArgs) → WebdavFile[source]#: Reopen file when unpickled.

webdav4.fsspec.translate_exceptions() → Iterator[None][source]#: Translate exceptions from Client to the fsspec compatible one.

webdav4.fsspec.translate_info(item: Union[str, Dict[str, Any]]) → Dict[str, Any][source]#: Translate info from the client to as per fsspec requirements.