Skip to main content

Asyncio WebDAV 客户端,基于原始包 https://github.com/ezhov-evgeny/webdav-client-python-3 但使用 aiohttp 而不是 requests

项目描述

aiowebdav

皮皮 Python 执行 车轮 行动

基于https://github.com/designerror/webdav-client-python打包 aiowebdav,但aiohttp使用requests. 它提供了使用 WebDAV 服务器的简单方法。

安装

$ pip install aiowebdav

示例使用

import asyncio
from aiowebdav.client import Client

options = {
 'webdav_hostname': "https://webdav.server.ru",
 'webdav_login': "login",
 'webdav_password': "password",
 "disable_check": True
}

async def main():
    client = Client(options)
    client.verify = False  # To not check SSL certificates (Default = True)
    client.execute_request("mkdir", 'directory_name')
asyncio.run(main())

Webdav API

Webdav API 是一组使用云存储的 webdav 操作。该集合包括以下动作: check, free, info, list, mkdir, clean, copy, move, download, upload,publishunpublish

配置客户端

必需的密钥是带有参数 name 的 WevDAV 服务器的主机名或 IP 地址webdav_hostname
对于 WebDAV 服务器中的身份验证,请使用webdav_login, webdav_password.
对于匿名登录,不要指定 auth 属性。

from aiowebdav.client import Client

options = {
 'webdav_hostname': "https://webdav.server.ru",
 'webdav_login': "login",
 'webdav_password': "password"
}
client = Client(options)

如果您的服务器不支持HEAD方法或有其他原因覆盖默认 WebDAV 方法的操作,请使用字典选项webdav_override_methods。密钥应在以下列表中:check, free, info, list, mkdir, clean, copy, move, download, upload, publishunpublish. 该值应该是 WebDAV 方法的字符串名称,例如GET.

from aiowebdav.client import Client

options = {
 'webdav_hostname': "https://webdav.server.ru",
 'webdav_login': "login",
 'webdav_password': "password",
 'webdav_override_methods': {
  'check': 'GET'
 }

}
client = Client(options)

要配置请求超时,您可以使用webdav_timeout以秒为单位的 int 值选项,默认情况下超时设置为 30 秒。

from aiowebdav.client import Client

options = {
 'webdav_hostname': "https://webdav.server.ru",
 'webdav_login': "login",
 'webdav_password': "password",
 'webdav_timeout': 30
}
client = Client(options)

当一个代理服务器你需要指定设置来通过它进行连接。

from aiowebdav.client import Client

options = {
 'webdav_hostname': "https://webdav.server.ru",
 'webdav_login': "w_login",
 'webdav_password': "w_password",
 'webdav_proxy': "http://127.0.0.1:8080",
 'webdav_proxy_auth': "xxx",
}
client = Client(options)

如果要使用证书路径,证书和私钥定义如下:

from aiowebdav.client import Client

options = {
 'webdav_hostname': "https://webdav.server.ru",
 'webdav_login': "w_login",
 'webdav_password': "w_password",
 'webdav_ssl': 'sslcontext'
}
client = Client(options)

或者您想限制速度或打开详细模式:

options = {
 ...
 'recv_speed' : 3000000,
 'send_speed' : 3000000,
 'verbose'    : True
}
client = Client(options)

recv_speed:速率限制数据下载速度,以每秒字节数为单位。默认为无限速度。
send_speed:速率限制数据上传速度,以每秒字节数为单位。默认为无限速度。
详细:设置详细模式开/关。默认情况下详细模式是关闭的。

此外,如果您的服务器不支持check,可以禁用它:

options = {
 ...
 'disable_check': True
}
client = Client(options)

默认情况下,启用远程资源检查。

用于配置内容下载的块大小使用chunk_size参数,默认为65536

options = {
 ...
 'chunk_size': 65536
}
client = Client(options)

异步方法

# Checking existence of the resource

await client.check("dir1/file1")
await client.check("dir1")
# Get information about the resource

await client.info("dir1/file1")
await client.info("dir1/")
# Check free space

free_size = await client.free()
# Get a list of resources

files1 = await client.list()
files2 = await client.list("dir1")
files3 = await client.list("dir1", get_info=True) # returns a list of dictionaries with files details
# Create directory

await client.mkdir("dir1/dir2")
# Delete resource

await client.clean("dir1/dir2")
# Copy resource

await client.copy(remote_path_from="dir1/file1", remote_path_to="dir2/file1")
await client.copy(remote_path_from="dir2", remote_path_to="dir3")
# Move resource

await client.move(remote_path_from="dir1/file1", remote_path_to="dir2/file1")
await client.move(remote_path_from="dir2", remote_path_to="dir3")
# Download a resource

await client.download(remote_path="dir1/file1", local_path="~/Downloads/file1")
await client.download(remote_path="dir1/dir2/", local_path="~/Downloads/dir2/")
# Upload resource

await client.upload(remote_path="dir1/file1", local_path="~/Documents/file1")
await client.upload(remote_path="dir1/dir2/", local_path="~/Documents/dir2/")
# Publish the resource

link = await client.publish("dir1/file1")
link = await client.publish("dir2")
# Unpublish resource

await client.unpublish("dir1/file1")
await client.unpublish("dir2")
# Exception handling

from aiowebdav.exceptions import WebDavException

try:
 ...
except WebDavException as exception:
...
# Get the missing files

await client.pull(remote_directory='dir1', local_directory='~/Documents/dir1')
# Send missing files

await client.push(remote_directory='dir1', local_directory='~/Documents/dir1')
# Unload resource

kwargs = {
 'remote_path': "dir1/file1",
 'local_path':  "~/Downloads/file1",
 'callback':    callback
}
client.upload_async(**kwargs)

kwargs = {
 'remote_path': "dir1/dir2/",
 'local_path':  "~/Downloads/dir2/",
 'callback':    callback
}
client.upload_async(**kwargs)

资源 API

使用 OOP 概念的资源 API,可实现云级资源。

# Get a resource

res1 = client.resource("dir1/file1")
# Work with the resource

await res1.rename("file2")
await res1.move("dir1/file2")
await res1.copy("dir2/file1")
info = await res1.info()
await res1.read_from(buffer)
await res1.read(local_path="~/Documents/file1")
await res1.write_to(buffer)
await res1.write(local_path="~/Downloads/file1")

对于贡献者

准备开发环境

  1. 在你的开发机器上安装 docker
  2. 通过以下命令从项目的根文件夹启动 WebDAV 服务器进行测试,或者在第二个命令中将路径更改为confdir 以进行更正:
docker pull bytemark/webdav
docker run -d --name webdav -e AUTH_TYPE=Basic -e USERNAME=alice -e PASSWORD=secret1234 -v conf:/usr/local/apache2/conf -p 8585:80 bytemark/webdav

代码约定

请根据 PEP8 样式指南检查您的代码。

运行测试

  1. 检查 webdav 容器是否已在本地计算机上启动
  2. 在项目的根文件夹中执行以下命令:
python -m unittest discover -s tests

准备一个拉取请求

请在创建 PR 之前使用此检查表:

  1. 您的代码应根据 PEP8 格式化
  2. 所有测试都应该成功通过
  3. 您的更改不应更改以前的默认行为,排除缺陷
  4. 测试涵盖所有更改

项目详情


下载文件

下载适用于您平台的文件。如果您不确定要选择哪个,请了解有关安装包的更多信息。

源分布

aiowebdav-0.1.0rc5.tar.gz (24.0 kB 查看哈希

已上传 source

内置分布

aiowebdav-0.1.0rc5-py3-none-any.whl (22.6 kB 查看哈希

已上传 py3