您现在的位置是：首页 > Python

当前栏目

乾坤大挪移,如何将同步阻塞(sync)三方库包转换为异步非阻塞(async)模式？Python3.10实现。

2023-02-19 12:19:43 时间

众所周知，异步并发编程可以帮助程序更好地处理阻塞操作，比如网络 IO 操作或文件 IO 操作，避免因等待这些操作完成而导致程序卡住的情况。云存储文件传输场景正好包含网络 IO 操作和文件 IO 操作，比如业内相对著名的七牛云存储，官方sdk的默认阻塞传输模式虽然差强人意，但未免有些循规蹈矩，不够锐意创新。在全球同性交友网站Github上找了一圈，也没有找到异步版本，那么本次我们来自己动手将同步阻塞版本改造为异步非阻塞版本，并上传至Python官方库。

异步改造

首先参见七牛云官方接口文档：https://developer.qiniu.com/kodo，新建qiniu_async.py文件：

# @Author:Liu Yue (v3u.cn)
# @Software:Vscode
# @Time:2022/12/30

import base64
import hmac
import time
from hashlib import sha1
import json
import httpx
import aiofiles



class Qiniu:

    def __init__(self, access_key, secret_key):
        """初始化"""
        self.__checkKey(access_key, secret_key)
        self.__access_key = access_key
        self.__secret_key = secret_key.encode('utf-8')

    def get_access_key(self):
        return self.__access_key

    def get_secret_key(self):
        return self.__secret_key

    def __token(self, data):
        hashed = hmac.new(self.__secret_key,data.encode('utf-8'), sha1)
        return self.urlsafe_base64_encode(hashed.digest())

    def token(self, data):
        return '{0}:{1}'.format(self.__access_key, self.__token(data))

    def token_with_data(self, data):
        data = self.urlsafe_base64_encode(data)
        return '{0}:{1}:{2}'.format(
            self.__access_key, self.__token(data), data)

    def urlsafe_base64_encode(self,data):

        if isinstance(data, str):
            data = data.encode('utf-8')

        ret = base64.urlsafe_b64encode(data)

        data = ret.decode('utf-8')

        return data


    @staticmethod
    def __checkKey(access_key, secret_key):
        if not (access_key and secret_key):
            raise ValueError('invalid key')


    def upload_token(
            self,
            bucket,
            key=None,
            expires=3600,
            policy=None,
            strict_policy=True):
        """生成上传凭证

        Args:
            bucket:  上传的空间名
            key:     上传的文件名，默认为空
            expires: 上传凭证的过期时间，默认为3600s
            policy:  上传策略，默认为空

        Returns:
            上传凭证
        """
        if bucket is None or bucket == '':
            raise ValueError('invalid bucket name')

        scope = bucket
        if key is not None:
            scope = '{0}:{1}'.format(bucket, key)

        args = dict(
            scope=scope,
            deadline=int(time.time()) + expires,
        )

        return self.__upload_token(args)

    @staticmethod
    def up_token_decode(up_token):
        up_token_list = up_token.split(':')
        ak = up_token_list[0]
        sign = base64.urlsafe_b64decode(up_token_list[1])
        decode_policy = base64.urlsafe_b64decode(up_token_list[2])
        decode_policy = decode_policy.decode('utf-8')
        dict_policy = json.loads(decode_policy)
        return ak, sign, dict_policy

    def __upload_token(self, policy):
        data = json.dumps(policy, separators=(',', ':'))
        return self.token_with_data(data)


    @staticmethod
    def __copy_policy(policy, to, strict_policy):
        for k, v in policy.items():
            if (not strict_policy) or k in _policy_fields:
                to[k] = v

这里有两个很关键的异步非阻塞三方库，分别是httpx和aiofiles，对应处理网络IO和文件IO阻塞问题：

pip3 install httpx
pip3 install aiofiles

随后按照文档流程通过加密方法获取文件上传token，这里无须进行异步改造，因为并不涉及IO操作：

q = Qiniu(access_key,access_secret)

token = q.upload_token("空间名称")

print(token)

程序返回：

➜  mydemo git:(master) ✗ /opt/homebrew/bin/python3.10 "/Users/liuyue/wodfan/work/mydemo/src/test.py"
q06bq54Ps5JLfZyP8Ax-qvByMBdu8AoIVJpMco2m:8RjIo9a4CxHM3009DwjbMxDzlU8=:eyJzY29wZSI6ImFkLWgyMTEyIiwiZGVhZGxpbmUiOjE2NzIzNjg2NTd9

接着添加文件流推送方法，先看官方原版逻辑：

def put_data(
        up_token, key, data, params=None, mime_type='application/octet-stream', check_crc=False, progress_handler=None,
        fname=None, hostscache_dir=None, metadata=None):
    """上传二进制流到七牛
    Args:
        up_token:         上传凭证
        key:              上传文件名
        data:             上传二进制流
        params:           自定义变量，规格参考 https://developer.qiniu.com/kodo/manual/vars#xvar
        mime_type:        上传数据的mimeType
        check_crc:        是否校验crc32
        progress_handler: 上传进度
        hostscache_dir:   host请求 缓存文件保存位置
        metadata:         元数据
    Returns:
        一个dict变量，类似 {"hash": "<Hash string>", "key": "<Key string>"}
        一个ResponseInfo对象
    """
    final_data = b''
    if hasattr(data, 'read'):
        while True:
            tmp_data = data.read(config._BLOCK_SIZE)
            if len(tmp_data) == 0:
                break
            else:
                final_data += tmp_data
    else:
        final_data = data

    crc = crc32(final_data)
    return _form_put(up_token, key, final_data, params, mime_type,
                     crc, hostscache_dir, progress_handler, fname, metadata=metadata)

def _form_put(up_token, key, data, params, mime_type, crc, hostscache_dir=None, progress_handler=None, file_name=None,
              modify_time=None, keep_last_modified=False, metadata=None):
    fields = {}
    if params:
        for k, v in params.items():
            fields[k] = str(v)
    if crc:
        fields['crc32'] = crc
    if key is not None:
        fields['key'] = key

    fields['token'] = up_token
    if config.get_default('default_zone').up_host:
        url = config.get_default('default_zone').up_host
    else:
        url = config.get_default('default_zone').get_up_host_by_token(up_token, hostscache_dir)
    # name = key if key else file_name

    fname = file_name
    if not fname or not fname.strip():
        fname = 'file_name'

    # last modify time
    if modify_time and keep_last_modified:
        fields['x-qn-meta-!Last-Modified'] = rfc_from_timestamp(modify_time)

    if metadata:
        for k, v in metadata.items():
            if k.startswith('x-qn-meta-'):
                fields[k] = str(v)

    r, info = http._post_file(url, data=fields, files={'file': (fname, data, mime_type)})
    if r is None and info.need_retry():
        if info.connect_failed:
            if config.get_default('default_zone').up_host_backup:
                url = config.get_default('default_zone').up_host_backup
            else:
                url = config.get_default('default_zone').get_up_host_backup_by_token(up_token, hostscache_dir)
        if hasattr(data, 'read') is False:
            pass
        elif hasattr(data, 'seek') and (not hasattr(data, 'seekable') or data.seekable()):
            data.seek(0)
        else:
            return r, info
        r, info = http._post_file(url, data=fields, files={'file': (fname, data, mime_type)})

    return r, info

这里官方使用两个方法，先试用put_data方法将字符串转换为二进制文件流，随后调用_form_put进行同步上传操作，这里_form_put这个私有方法是可复用的，既兼容文件流也兼容文件实体，写法上非常值得我们借鉴，弄明白了官方原版的流程后，让我们撰写文件流传输的异步版本：

# 上传文件流
    async def upload_data(self,up_token, key,data,url="http://up-z1.qiniup.com",params=None,mime_type='application/octet-stream',file_name=None,metadata=None):

        data.encode('utf-8')
        
        fields = {}
        if params:
            for k, v in params.items():
                fields[k] = str(v)

        if key is not None:
            fields['key'] = key
        fields['token'] = up_token

        fname = file_name
        if not fname or not fname.strip():
            fname = 'file_name'

        async with httpx.AsyncClient() as client:

            # 调用异步使用await关键字
            res = await client.post(url,data=fields,files={'file': (fname,data,mime_type)})

            print(res.text)

这里我们声明异步方法upload_data，通过encode直接转换文件流，并使用异步httpx.AsyncClient()对象将文件流推送到官网接口地址：up-z1.qiniup.com

随后进行测试：

import asyncio
q = qiniu_async.Qiniu("accesskey","accesssecret")

token = q.upload_token("空间名称")

#文件流上传
asyncio.run(q.upload_data(token,"3343.txt","123测试"))

程序返回：

➜  mydemo git:(master) ✗ /opt/homebrew/bin/python3.10 "/Users/liuyue/wodfan/work/mydemo/src/test.py"
{"hash":"FtnQXAXft5AsOH1mrmXGaRzSt-95","key":"33434.txt"}

接口会返回文件流的hash编码，没有问题。

接着查看文件上传流程：

def put_file(up_token, key, file_path, params=None,
             mime_type='application/octet-stream', check_crc=False,
             progress_handler=None, upload_progress_recorder=None, keep_last_modified=False, hostscache_dir=None,
             part_size=None, version=None, bucket_name=None, metadata=None):
    """上传文件到七牛
    Args:
        up_token:                 上传凭证
        key:                      上传文件名
        file_path:                上传文件的路径
        params:                   自定义变量，规格参考 https://developer.qiniu.com/kodo/manual/vars#xvar
        mime_type:                上传数据的mimeType
        check_crc:                是否校验crc32
        progress_handler:         上传进度
        upload_progress_recorder: 记录上传进度，用于断点续传
        hostscache_dir:           host请求 缓存文件保存位置
        version:                  分片上传版本 目前支持v1/v2版本 默认v1
        part_size:                分片上传v2必传字段 默认大小为4MB 分片大小范围为1 MB - 1 GB
        bucket_name:              分片上传v2字段 空间名称
        metadata:                 元数据信息
    Returns:
        一个dict变量，类似 {"hash": "<Hash string>", "key": "<Key string>"}
        一个ResponseInfo对象
    """
    ret = {}
    size = os.stat(file_path).st_size
    with open(file_path, 'rb') as input_stream:
        file_name = os.path.basename(file_path)
        modify_time = int(os.path.getmtime(file_path))
        if size > config.get_default('default_upload_threshold'):
            ret, info = put_stream(up_token, key, input_stream, file_name, size, hostscache_dir, params,
                                   mime_type, progress_handler,
                                   upload_progress_recorder=upload_progress_recorder,
                                   modify_time=modify_time, keep_last_modified=keep_last_modified,
                                   part_size=part_size, version=version, bucket_name=bucket_name, metadata=metadata)
        else:
            crc = file_crc32(file_path)
            ret, info = _form_put(up_token, key, input_stream, params, mime_type,
                                  crc, hostscache_dir, progress_handler, file_name,
                                  modify_time=modify_time, keep_last_modified=keep_last_modified, metadata=metadata)
    return ret, info

这里官方使用的是标准库上下文管理器同步读取文件，改写为异步方法：

# 上传文件实体
    async def upload_file(self,up_token,key,path,url="http://up-z1.qiniup.com",params=None,mime_type='application/octet-stream',file_name=None,metadata=None):


        async with aiofiles.open(path, mode='rb') as f:
            contents = await f.read()
        
        fields = {}
        if params:
            for k, v in params.items():
                fields[k] = str(v)

        if key is not None:
            fields['key'] = key
        fields['token'] = up_token

        fname = file_name
        if not fname or not fname.strip():
            fname = 'file_name'

        async with httpx.AsyncClient() as client:

            # 调用异步使用await关键字
            res = await client.post(url,data=fields,files={'file': (fname,contents,mime_type)})

            print(res.text)

通过aiofiles异步读取文件后，在通过httpx.AsyncClient()进行异步传输。

需要注意的是，这里默认传输到up-z1.qiniup.com接口，如果是不同区域的云存储服务器，需要更改url参数的值，具体服务器接口列表请参照官网文档。

至此，文件流和文件异步传输就改造好了。

上传至Python官方库

为了方便广大七牛云用户使用异步传输版本库，可以将qiniu-async上传到Python官方库，首先注册成为Python官方库的开发者：pypi.org/

随后在项目根目录下新建setup.py文件：

import setuptools
import pathlib

here = pathlib.Path(__file__).parent.resolve()
long_description = (here / "README.md").read_text(encoding="utf-8")

setuptools.setup(
    name="qiniu-async",
    version="1.0.1",
    author="LiuYue",
    author_email="zcxey2911@gmail.com",
    description="qiniu_async python library",
    long_description=long_description,
    long_description_content_type="text/markdown",
    url="https://github.com/qiniu-async",
    packages=setuptools.find_packages(),
    license="Apache 2.0",
    classifiers=[
        "Development Status :: 3 - Alpha",
        "Intended Audience :: Developers",
        "Programming Language :: Python :: 3",
        "Programming Language :: Python :: 3.7",
        "Programming Language :: Python :: 3.8",
        "Programming Language :: Python :: 3.9",
        "Programming Language :: Python :: 3.10",
        "Programming Language :: Python :: 3 :: Only",
        "License :: OSI Approved :: MIT License",
        "Operating System :: OS Independent",

    ],
    keywords="qiniu, qiniu_async, async",
    py_modules=[
        'qiniu_async'
    ],
    install_requires=["aiofiles","httpx"],
)

这是安装文件，主要为了声明该模块的名称、作者、版本以及依赖库。

随后本地打包文件：

python3 setup.py sdist

程序会根据setup.py文件生成压缩包：

➜  qiniu_async tree
.
├── README.md
├── dist
│   └── qiniu-async-1.0.1.tar.gz
├── https:
│   └── github.com
│       └── zcxey2911
│           └── qiniu-async.git
├── qiniu_async.egg-info
│   ├── PKG-INFO
│   ├── SOURCES.txt
│   ├── dependency_links.txt
│   ├── requires.txt
│   └── top_level.txt
├── qiniu_async.py
└── setup.py

接着安装twine库, 准备提交Python官网：

pip3 install twine

随后在根目录运行命令提交：

twine upload dist/*

在官网进行查看：https://pypi.org/project/qiniu-async/

随后本地就可以直接通过pip命令句进行安装了：

pip install qiniu-async -i https://pypi.org/simple

非常方便。

结语

云端存储，异步加持，猛虎添翼，未敢拥钵独飨，除了通过pip安装qiniu-async库，也奉上Github项目地址：https://github.com/zcxey2911/qiniu-async ，与众乡亲同飨。

猜你喜欢

最长无重复子串
写技术博客的一些心得分享
Java 多线程（七）：线程池
Java 多线程（五）：锁（三）
Java 多线程（四）：锁（二）
Java 多线程（三）：锁（一）
Java 多线程（二）：并发编程的三大特性
线性时间非比较类排序
Java 多线程（一）：基础
合并k个已排序的链表
HDFS 高可用分布式环境搭建
合并两个有序数组
连续子数组的最大和
HDFS 分布式环境搭建
容器盛水问题
大数加法
HDFS 伪分布式环境搭建
设计LRU缓存结构
两数之和
使用单调栈来解决的一些问题

zl程序教程

当前栏目

乾坤大挪移,如何将同步阻塞(sync)三方库包转换为异步非阻塞(async)模式？Python3.10实现。

异步改造

上传至Python官方库

结语

相关文章