api

chunked_upload(nc_client, local_source_file, remote_path, chunk_size=_DEFAULT_CHUNK_SIZE, keep_mtime=True)

Upload file to remote by creating a folder containing chunks if needed.

Chunks are enumerated from 000000000000001 until the final chunk. The folder will be named remote_path + _chunked. If the file is smaller than the chunk_size, will try to upload file in one piece.

Parameters:

Name	Type	Description	Default
nc_client	Client	Client with shared folder	required
local_source_file	Path	Local file to upload	required
remote_path	str	Path where file be stored remotely.	required
chunk_size	int	Chunk size in bytes.	_DEFAULT_CHUNK_SIZE
keep_mtime	bool	If True sends modified time in header.	True

Returns:

Type	Description
bool	True if successful

Raises:

Type	Description
ServerError	owncloud.owncloud.HTTPResponseError if server throws error

Source code in /home/docs/checkouts/readthedocs.org/user_builds/nephelai/envs/stable/lib/python3.10/site-packages/nephelai/api.py

def chunked_upload(
    nc_client: owncloud.Client,
    local_source_file: pathlib.Path,
    remote_path: str,
    chunk_size: int = _DEFAULT_CHUNK_SIZE,
    keep_mtime: bool = True,
) -> bool:
    """Upload file to remote by creating a folder containing chunks if needed.

    Chunks are enumerated from 000000000000001 until the final chunk.
    The folder will be named `remote_path` + `_chunked`.
    If the file is smaller than the `chunk_size`, will try to upload
    file in one piece.

    Args:
        nc_client: Client with shared folder
        local_source_file: Local file to upload
        remote_path: Path where file be stored remotely.
        chunk_size: Chunk size in bytes.
        keep_mtime: If True sends modified time in header.

    Returns:
        True if successful

    Raises:
        ServerError: owncloud.owncloud.HTTPResponseError if server throws error
    """
    result = True

    remote_path = nc_client._normalize_path(remote_path)
    if remote_path.endswith("/"):
        remote_path += os.path.basename(local_source_file)

    original_remote_path = remote_path
    create_nc_folders(
        my_dir=pathlib.Path(original_remote_path).parent, nc_client=nc_client
    )
    remote_path += CHUNKED_SUFFIX + "/"

    stat_result = os.stat(local_source_file)

    file_handle = open(local_source_file, "rb", 8192)
    file_handle.seek(0, os.SEEK_END)
    size = file_handle.tell()

    if size <= chunk_size:
        return nc_client.put_file(
            remote_path=original_remote_path,
            local_source_file=local_source_file,
            chunked=False,
        )

    file_handle.seek(0)

    nc_client.mkdir(remote_path)
    headers = {}
    if keep_mtime:
        headers["X-OC-MTIME"] = str(int(stat_result.st_mtime))

    if size == 0:
        return nc_client._make_dav_request(
            "PUT", original_remote_path, data="", headers=headers
        )

    try:
        chunk_count = int(math.ceil(float(size) / float(chunk_size)))
        for chunk_index in trange(0, int(chunk_count), desc="Uploading chunks"):
            data = file_handle.read(chunk_size)
            chunk_name = f"{remote_path}{chunk_index:015d}"
            if not nc_client._make_dav_request(
                "PUT", chunk_name, data=data, headers=headers
            ):
                result = False
                break
    except owncloud.owncloud.HTTPResponseError as ServerError:
        nc_client.delete(remote_path)
        file_handle.close()
        raise ServerError
    file_handle.close()
    return result

create_nc_folders(my_dir, nc_client)

Create nextcloud folders using given owncloud client.

Will create all subpaths if needed.

Parameters:

Name	Type	Description	Default
my_dir	Path	Full path to create	required
nc_client	Client	Client with shared folder	required

Returns:

Type	Description
bool	True if folders did not exist, False otherwise

Source code in /home/docs/checkouts/readthedocs.org/user_builds/nephelai/envs/stable/lib/python3.10/site-packages/nephelai/api.py

def create_nc_folders(my_dir: pathlib.Path, nc_client: owncloud.Client) -> bool:
    """Create nextcloud folders using given owncloud client.

    Will create all subpaths if needed.

    Args:
        my_dir: Full path to create
        nc_client: Client with shared folder

    Returns:
        True if folders did not exist, False otherwise
    """
    accumulated_part = ""
    had_to_create = False
    for dir_part in my_dir.parts:
        accumulated_part = os.path.join(accumulated_part, dir_part)
        try:
            nc_client.file_info(accumulated_part)
        except owncloud.owncloud.HTTPResponseError as oc_error:
            if oc_error.status_code == 404:
                nc_client.mkdir(accumulated_part)
                had_to_create = True
    return had_to_create

download(remote_path, local_path=None)

Download file from remote.

If the file is chunked it will be reconstructed while downloading. If no local path is given it will be written in the current directory. If it exists it will be replaced.

Parameters:

Name	Type	Description	Default
remote_path	str	Path where the file lies in the shared folder	required
local_path	Optional[str]	Path where file should downloaded to	None

Returns:

Type	Description
Optional[bool]	True if successful, else None

Source code in /home/docs/checkouts/readthedocs.org/user_builds/nephelai/envs/stable/lib/python3.10/site-packages/nephelai/api.py

def download(remote_path: str, local_path: Optional[str] = None) -> Optional[bool]:
    """Download file from remote.

    If the file is chunked it will be reconstructed while downloading. If no local path is given it will be written
    in the current directory. If it exists it will be replaced.

    Args:
        remote_path: Path where the file lies in the shared folder
        local_path: Path where file should downloaded to

    Returns:
        True if successful, else None
    """
    oc = get_oc()
    if local_path is None:
        local_path = pathlib.Path(remote_path).name
    if not remote_path.startswith("/"):
        remote_path = "/" + remote_path
    file_state, file_info = _check_file_state(oc=oc, remote_path=remote_path)
    if file_state == FileStateDoesNotExist:
        return None
    if file_state == FileStateIsDirUnchunked:
        assert file_info  # for mypy
        remote_dir_name = pathlib.Path(file_info.path).name
        if not local_path.endswith(remote_dir_name):
            local_path = os.path.join(local_path, remote_dir_name)
    return _download(
        oc=oc,
        relative_path="",
        remote_base=remote_path,
        local_base=local_path,
        file_state=file_state,
        file_info=file_info,
    )

upload(file_to_upload, nc_path, chunk_size='100MiB', debug=False)

Upload to password protected shared folder.

If file is larger than chunk_size a folder will be created, and the chunks will be uploaded there.

Parameters:

Name	Type	Description	Default
file_to_upload	Union[Path, str]	The file to upload	required
nc_path	str	The remote path	required
chunk_size	Union[int, str]	Chunk size as bytes or as human readable string	'100MiB'
debug	bool	If True, show debug info	False

Returns:

Type	Description
Optional[bool]	True if successful, else None

Source code in /home/docs/checkouts/readthedocs.org/user_builds/nephelai/envs/stable/lib/python3.10/site-packages/nephelai/api.py

def upload(
    file_to_upload: Union[pathlib.Path, str],
    nc_path: str,
    chunk_size: Union[int, str] = "100MiB",
    debug: bool = False,
) -> Optional[bool]:
    """Upload to password protected shared folder.

    If file is larger than `chunk_size` a folder will be created,
    and the chunks will be uploaded there.

    Args:
        file_to_upload: The file to upload
        nc_path: The remote path
        chunk_size: Chunk size as bytes or as human readable string
        debug: If True, show debug info

    Returns:
        True if successful, else None
    """
    if not isinstance(file_to_upload, pathlib.Path):
        file_to_upload = pathlib.Path(file_to_upload)
    if not isinstance(chunk_size, int):
        try:
            chunk_size = int(chunk_size)
        except ValueError:
            chunk_size = int(bitmath.parse_string(chunk_size).bytes)
    oc = get_oc()
    if file_to_upload.is_dir():
        create_nc_folders(file_to_upload, oc)
        for subfile in file_to_upload.iterdir():
            upload(
                subfile,
                os.path.join(nc_path, subfile.name),
                chunk_size=chunk_size,
                debug=debug,
            )
    else:
        chunked_upload(oc, file_to_upload, nc_path, chunk_size=chunk_size)
    return True