wcpan.drive.core.lib

File Transfers

async wcpan.drive.core.lib.upload_file_from_local(drive, path, parent, *, name=None, mime_type=None, media_info=None, timeout=None)

Upload a local file to the cloud drive.

Reads a local file and uploads it to the specified parent directory in the cloud. Supports resumable uploads with automatic retry on timeout, making it suitable for large files or unstable connections.

Parameters:

  • drive (Drive) – The Drive instance to upload to.

  • path (Path) – Absolute path to the local file to upload.

  • parent (Node) – Parent directory Node where file will be uploaded.

  • name (str | None) – Name for the uploaded file. If None, uses local filename.

  • mime_type (str | None) – MIME type for the file. If None, uses ‘application/octet-stream’.

  • media_info (MediaInfo | None) – Optional MediaInfo for images/videos.

  • timeout (float | None) – Timeout in seconds for each I/O operation. None means no timeout. If an operation times out, upload automatically resumes from the last successful position.

Returns:

The newly created Node representing the uploaded file.

Raises:
Return type:

Node

Example

Basic upload:

>>> from pathlib import Path, PurePath
>>> root = await drive.get_root()
>>> node = await upload_file_from_local(
...     drive,
...     Path("/home/user/document.pdf"),
...     root,
... )
>>> print(f"Uploaded: {node.name} ({node.size} bytes)")

Upload with custom name:

>>> node = await upload_file_from_local(
...     drive,
...     Path("/tmp/temp_file.dat"),
...     parent,
...     name="final_name.dat",
... )

Upload image with metadata:

>>> from wcpan.drive.core.types import MediaInfo
>>> media = MediaInfo.image(width=1920, height=1080)
>>> node = await upload_file_from_local(
...     drive,
...     Path("/photos/vacation.jpg"),
...     parent,
...     mime_type="image/jpeg",
...     media_info=media,
... )

Upload with timeout for resilience:

>>> node = await upload_file_from_local(
...     drive,
...     Path("/large_file.zip"),
...     parent,
...     timeout=30.0,  # 30 second timeout per chunk
... )

Note

The function automatically handles upload resumption on timeout. File reading also has timeout protection for unstable filesystems.

async wcpan.drive.core.lib.download_file_to_local(drive, node, path, *, timeout=None)

Download a cloud file to a local directory.

Downloads a file from the cloud to the specified local directory. Supports resumable downloads with automatic retry on timeout, making it suitable for large files or unstable connections.

If the download is interrupted, the next call will resume from where it left off using a temporary file with .__tmp__ suffix.

Parameters:

  • drive (Drive) – The Drive instance to download from.

  • node (Node) – The file Node to download.

  • path (Path) – Local directory path where file will be saved.

  • timeout (float | None) – Timeout in seconds for each I/O operation. None means no timeout. If an operation times out, download automatically resumes from the last successful position.

Returns:

Path to the downloaded file (path / node.name).

Raises:
Return type:

Path

Example

Basic download:

>>> from pathlib import Path, PurePath
>>> node = await drive.get_node_by_path(PurePath("/document.pdf"))
>>> local_path = await download_file_to_local(
...     drive,
...     node,
...     Path("/downloads"),
... )
>>> print(f"Downloaded to: {local_path}")

Download with timeout for resilience:

>>> local_path = await download_file_to_local(
...     drive,
...     node,
...     Path("/downloads"),
...     timeout=30.0,  # 30 second timeout per chunk
... )

Handling existing files:

>>> dest_dir = Path("/downloads")
>>> local_path = await download_file_to_local(drive, node, dest_dir)
>>> # If file already exists, returns existing path without re-downloading

Resuming interrupted download:

>>> # First attempt (interrupted)
>>> try:
...     local_path = await download_file_to_local(drive, node, dest_dir)
... except asyncio.TimeoutError:
...     print("Download interrupted")
>>>
>>> # Second attempt (resumes automatically)
>>> local_path = await download_file_to_local(drive, node, dest_dir)
>>> # Continues from where it left off using .tmp file

Note

  • If the file already exists and is complete, it’s returned immediately without re-downloading

  • Interrupted downloads leave a .__tmp__ file that’s used to resume

  • Empty files (size 0) are created without downloading

  • The function handles both filesystem and network timeouts

Node Operations

async wcpan.drive.core.lib.move_node(drive, src_path, dst_path)

Move or rename a node using path-based addressing.

Provides flexible path-based node movement with support for: - Absolute destination paths - Relative destination paths - Renaming in place - Moving to parent directory (..)

The function intelligently handles different path types and resolves them to appropriate Drive.move() calls.

Parameters:

  • drive (Drive) – The Drive instance to operate on.

  • src_path (PurePath) – Absolute path to the source node.

  • dst_path (PurePath) – Destination path (absolute or relative to source parent).

Returns:

The moved/renamed Node with updated properties.

Raises:

  • NodeNotFoundError – If source path doesn’t exist or destination parent doesn’t exist.

  • NodeExistsError – If destination is an existing file (won’t overwrite).

  • ValueError – If there’s no valid path to the destination parent.

Return type:

Node

Example

Renaming a file:

>>> from pathlib import PurePath
>>> moved = await move_node(
...     drive,
...     PurePath("/old_name.txt"),
...     PurePath("new_name.txt"),  # Relative path = rename
... )

Moving to different directory:

>>> moved = await move_node(
...     drive,
...     PurePath("/docs/file.txt"),
...     PurePath("/archive/file.txt"),  # Absolute path
... )

Moving to parent directory:

>>> moved = await move_node(
...     drive,
...     PurePath("/folder/subfolder/file.txt"),
...     PurePath(".."),  # Move up one level
... )

Moving and renaming:

>>> moved = await move_node(
...     drive,
...     PurePath("/docs/old.txt"),
...     PurePath("/archive/new.txt"),
... )

Note

  • If dst_path is “.”, the node is unchanged

  • If dst_path is absolute and points to a directory, the node is moved into that directory keeping its name

  • If dst_path is absolute and doesn’t exist, the node is moved to that exact path (parent directory must exist)

async wcpan.drive.core.lib.find_duplicate_nodes(drive, root_node=None)

Find nodes with duplicate names in the same directory.

Walks the directory tree and identifies any directories or files that share the same name within a single parent directory. This indicates filesystem inconsistencies that may need resolution.

Parameters:

  • drive (Drive) – The Drive instance to search.

  • root_node (Node | None) – Starting node for search. If None, uses drive root.

Returns:

List of duplicate groups, where each group is a list of Nodes with the same name in the same parent. Empty list if no duplicates.

Return type:

list[list[Node]]

Example

Finding duplicates:

>>> duplicates = await find_duplicate_nodes(drive)
>>> for group in duplicates:
...     print(f"Duplicate name '{group[0].name}':")
...     for node in group:
...         path = await drive.resolve_path(node)
...         print(f"  - {path} (ID: {node.id})")

Searching within a specific directory:

>>> docs = await drive.get_node_by_path(PurePath("/Documents"))
>>> duplicates = await find_duplicate_nodes(drive, docs)
>>> if not duplicates:
...     print("No duplicates found in Documents")

Note

Cloud storage services should prevent duplicates, but they can occur due to sync conflicts, API race conditions, or service bugs.

Path Utilities

wcpan.drive.core.lib.normalize_path(path)

Normalize an absolute path by resolving . and .. components.

Processes path segments to resolve relative references (. and ..) while preserving the absolute path structure. This is more robust than PurePath’s built-in normalization for cloud drive paths.

Parameters:

path (PurePath) – The absolute path to normalize.

Returns:

Normalized absolute PurePath with . and .. resolved.

Raises:

ValueError – If path is not absolute.

Return type:

PurePath

Example

Normalizing paths:

>>> from pathlib import PurePath
>>> path = PurePath("/docs/../photos/./image.jpg")
>>> normalized = normalize_path(path)
>>> print(normalized)
/photos/image.jpg

>>> path = PurePath("/a/b/c/../../d")
>>> print(normalize_path(path))
/a/d
wcpan.drive.core.lib.is_valid_name(name)

Check if a name is valid for files and directories.

Validates that a name doesn’t contain path separators or other invalid characters. Valid names are simple filenames without any path components.

Parameters:

name (str) – The name string to validate.

Returns:

True if the name is valid, False otherwise.

Return type:

bool

Example

Validating names:

>>> is_valid_name("document.txt")
True
>>> is_valid_name("my file.pdf")
True
>>> is_valid_name("path/to/file.txt")
False
>>> is_valid_name("file\name.txt")
False
>>> is_valid_name("./relative")
False

Change Helpers

wcpan.drive.core.lib.is_remove(change, /)

Check if a change action represents a node removal.

Type guard function that narrows ChangeAction to RemoveAction, enabling type-safe access to removal information.

Parameters:

change (ChangeAction) – The ChangeAction to check.

Returns:

True if the change is a RemoveAction, False otherwise.

Return type:

TypeGuard[RemoveAction]

Example

Using with type narrowing:

>>> async for change in drive.sync():
...     if is_remove(change):
...         removed, node_id = change  # Type is RemoveAction
...         print(f"Removed: {node_id}")

Filtering removed nodes:

>>> changes = []
>>> async for change in drive.sync():
...     changes.append(change)
>>> removed_ids = [node_id for c in changes if is_remove(c) for _, node_id in [c]]

See also

  • is_update(): Check if change is an UpdateAction

  • dispatch_change(): Pattern matching alternative

wcpan.drive.core.lib.is_update(change, /)

Check if a change action represents a node update or addition.

Type guard function that narrows ChangeAction to UpdateAction, enabling type-safe access to the updated Node.

Parameters:

change (ChangeAction) – The ChangeAction to check.

Returns:

True if the change is an UpdateAction, False otherwise.

Return type:

TypeGuard[UpdateAction]

Example

Using with type narrowing:

>>> async for change in drive.sync():
...     if is_update(change):
...         updated, node = change  # Type is UpdateAction
...         print(f"Updated: {node.name}")

Processing only updates:

>>> changes = []
>>> async for change in drive.sync():
...     changes.append(change)
>>> updated_nodes = [node for c in changes if is_update(c) for _, node in [c]]

See also

  • is_remove(): Check if change is a RemoveAction

  • dispatch_change(): Pattern matching alternative

wcpan.drive.core.lib.dispatch_change(change, /, *, on_remove, on_update)

Dispatch a change action to the appropriate handler function.

Provides pattern matching style dispatching for ChangeAction handling. Calls on_remove for removals or on_update for updates/additions.

Parameters:

  • change (ChangeAction) – The ChangeAction to dispatch.

  • on_remove (Callable[[str], R]) – Callback for RemoveAction. Receives the removed node’s ID.

  • on_update (Callable[[Node], R]) – Callback for UpdateAction. Receives the updated Node.

Returns:

The return value from whichever callback was invoked.

Return type:

R

Example

Basic dispatching:

>>> def handle_remove(node_id: str) -> None:
...     print(f"Removed: {node_id}")
...
>>> def handle_update(node: Node) -> None:
...     print(f"Updated: {node.name}")
...
>>> async for change in drive.sync():
...     dispatch_change(
...         change,
...         on_remove=handle_remove,
...         on_update=handle_update,
...     )

Collecting statistics:

>>> stats = {"removed": 0, "updated": 0}
>>> async for change in drive.sync():
...     dispatch_change(
...         change,
...         on_remove=lambda _: stats.update(removed=stats["removed"] + 1),
...         on_update=lambda _: stats.update(updated=stats["updated"] + 1),
...     )
>>> print(f"Changes: {stats['removed']} removed, {stats['updated']} updated")

Building index:

>>> index: dict[str, Node] = {}
>>> async for change in drive.sync():
...     dispatch_change(
...         change,
...         on_remove=lambda id_: index.pop(id_, None),
...         on_update=lambda node: index.update({node.id: node}),
...     )

See also

  • is_remove() and is_update(): Type guard alternatives

Miscellaneous

async wcpan.drive.core.lib.else_none(aw)

Execute an awaitable, returning None if NodeNotFoundError is raised.

Convenience wrapper for operations that may fail due to missing nodes. Converts NodeNotFoundError exceptions into None returns for cleaner error handling.

Parameters:

aw (Awaitable) – The awaitable to execute.

Returns:

The awaitable’s result if successful, None if NodeNotFoundError.

Return type:

T | None

Example

Checking if a path exists:

>>> node = await else_none(drive.get_node_by_path(path))
>>> if node is None:
...     print("Path does not exist")
... else:
...     print(f"Found: {node.name}")

Safe child lookup:

>>> child = await else_none(drive.get_child_by_name("file.txt", parent))
>>> if child:
...     await drive.delete(child)