Введение
YaDisk-async - это модифицированная версия YaDisk с поддержкой async/await
, использует aiohttp вместо requests.
Предупреждение
Данная библиотека больше не поддерживается, используйте yadisk>=2.0.0 вместо неё, она также поддерживает асинхронный API.
Использование библиотеки почти не отличается, не считая того, что закрывать все сессии нужно вручную (может быть сделано с помощью YaDisk.close
или через конструкцию async with
).
Установка
pip install yadisk-async
или
python setup.py install
Примеры
import yadisk_async
y = yadisk_async.YaDisk(token="<token>")
# or
# y = yadisk_async.YaDisk("<application-id>", "<application-secret>", "<token>")
# Check if the token is valid
print(await y.check_token())
# Get disk information
print(await y.get_disk_info())
# Print files and directories at "/some/path"
print([i async for i in await y.listdir("/some/path")])
# Upload "file_to_upload.txt" to "/destination.txt"
await y.upload("file_to_upload.txt", "/destination.txt")
# Same thing
with open("file_to_upload.txt", "rb") as f:
await y.upload(f, "/destination.txt")
# Same but with aiofiles
async with aiofiles.open("file_to_upload.txt", "rb") as f:
await y.upload(f, "/destination.txt")
# Download "/some-file-to-download.txt" to "downloaded.txt"
await y.download("/some-file-to-download.txt", "downloaded.txt")
# Same but with aiofiles
async with aiofiles.open("downloaded.txt", "wb") as f:
await y.download("/some-file-to-download.txt", f)
# Permanently remove "/file-to-remove"
await y.remove("/file-to-remove", permanently=True)
# Create a new directory at "/test-dir"
print(await y.mkdir("/test-dir"))
# Always remember to close all the connections or you'll get a warning
await y.close()
Получение токена через код подтверждения
import asyncio
import sys
import yadisk_async
async def main():
async with yadisk_async.YaDisk("application-id>", "<application-secret>") as y:
url = y.get_code_url()
print("Go to the following url: %s" % url)
code = input("Enter the confirmation code: ")
try:
response = await y.get_token(code)
except yadisk_async.exceptions.BadRequestError:
print("Bad code")
sys.exit(1)
y.token = response.access_token
if await y.check_token():
print("Sucessfully received token!")
else:
print("Something went wrong. Not sure how though...")
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
Рекурсивная загрузка файлов
import asyncio
import posixpath
import os
import yadisk_async
def recursive_upload(from_dir, to_dir, n_parallel_requests=5):
loop = asyncio.get_event_loop()
y = yadisk_async.YaDisk(token="<application-token>")
try:
async def upload_files(queue):
while queue:
in_path, out_path = queue.pop(0)
print("Uploading %s -> %s" % (in_path, out_path))
try:
await y.upload(in_path, out_path)
except yadisk_async.exceptions.PathExistsError:
print("%s already exists" % (out_path,))
async def create_dirs(queue):
while queue:
path = queue.pop(0)
print("Creating directory %s" % (path,))
try:
await y.mkdir(path)
except yadisk_async.exceptions.PathExistsError:
print("%s already exists" % (path,))
mkdir_queue = []
upload_queue = []
print("Creating directory %s" % (to_dir,))
try:
loop.run_until_complete(y.mkdir(to_dir))
except yadisk_async.exceptions.PathExistsError:
print("%s already exists" % (to_dir,))
for root, dirs, files in os.walk(from_dir):
rel_dir_path = root.split(from_dir)[1].strip(os.path.sep)
rel_dir_path = rel_dir_path.replace(os.path.sep, "/")
dir_path = posixpath.join(to_dir, rel_dir_path)
for dirname in dirs:
mkdir_queue.append(posixpath.join(dir_path, dirname))
for filename in files:
out_path = posixpath.join(dir_path, filename)
rel_dir_path_sys = rel_dir_path.replace("/", os.path.sep)
in_path = os.path.join(from_dir, rel_dir_path_sys, filename)
upload_queue.append((in_path, out_path))
tasks = [upload_files(upload_queue) for i in range(n_parallel_requests)]
tasks.extend(create_dirs(mkdir_queue) for i in range(n_parallel_requests))
loop.run_until_complete(asyncio.gather(*tasks))
finally:
loop.run_until_complete(y.close())
from_dir = input("Directory to upload: ")
to_dir = input("Destination directory: ")
recursive_upload(from_dir, to_dir, 5)
Задание пользовательских свойств файлов
import asyncio
import yadisk_async
async def main():
async with yadisk_async.YaDisk(token="<application-token>") as y:
path = input("Enter a path to patch: ")
properties = {"speed_of_light": 299792458,
"speed_of_light_units": "meters per second",
"message_for_owner": "MWAHAHA! Your file has been patched by an evil script!"}
meta = await y.patch(path, properties)
print("\nNew properties: ")
for k, v in meta.custom_properties.items():
print("%s: %r" % (k, v))
answer = input("\nWant to get rid of them? (y/[n]) ")
if answer.lower() in ("y", "yes"):
properties = {k: None for k in properties}
await y.patch(path, properties)
print("Everything's back as usual")
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
Очищение корзины
import asyncio
import sys
import yadisk_async
async def main():
async with yadisk_async.YaDisk(token="<application-token>") as y:
answer = input("Are you sure about this? (y/[n]) ")
if answer.lower() in ("y", "yes"):
print("Emptying the trash bin...")
operation = await y.remove_trash("/")
print("It might take a while...")
if operation is None:
print("Nevermind. The deed is done.")
sys.exit(0)
while True:
status = await y.get_operation_status(operation.href)
if status == "in-progress":
await asyncio.sleep(5)
print("Still waiting...")
elif status == "success":
print("Success!")
break
else:
print("Got some weird status: %r" % (status,))
print("That's not normal")
break
else:
print("Not going to do anything")
loop = asyncio.get_event_loop()
loop.run_until_complete(main())