一同养成写作习气！这是我参加「日新计划 4 月更文挑战」的第2天，点击检查活动详情。
大家好~我是
米洛！
我正在从0到1打造一个开源的接口测验平台, 也在编写一套与之对应的教程，期望大家多多支撑。
欢迎重视我的大众号米洛的测开日记，一同交流学习!
回顾
上一节咱们编写了工作台部分内容，期间修复了不少bug，也完善了很多特性。鉴于最近gitee不再供给图床功用了，所以咱们要自己切换到免费的七牛云存储之中。
话不多说，咱们接着肝。
修正装备
还记得咱们之前做过的装备文件吗？咱们把oss类型改为qiniu，并附上咱们申请好的token。
增加Config中关于七牛云oss的装备项
完善app/middleware/oss/qiniu.py
import os
from io import BytesIO
import aiohttp
from qiniu import Auth, put_stream, BucketManager
from app.middleware.oss import OssFile
from config import Config
class QiniuOss(OssFile):
_base_path = "pity"
def __init__(self, access_key_id: str, access_key_secret: str, bucket: str):
self.auth = Auth(access_key_id, access_key_secret)
self.bucket = bucket
self.bucket_manager = BucketManager(self.auth)
def get_full_path(self, filepath, base_path: str = None):
base_path = base_path if base_path is not None else self._base_path
return f"{base_path}/{filepath}"
@staticmethod
def _convert_to_stream(content):
stream = BytesIO()
stream.write(content)
return stream
async def create_file(self, filepath: str, content: bytes, base_path: str = None):
key = self.get_full_path(filepath, base_path)
token = self.auth.upload_token(self.bucket, key, 3600)
file_name = os.path.basename(filepath)
ret, info = put_stream(token, key, QiniuOss._convert_to_stream(content), file_name, len(content))
if ret['key'] != key:
raise Exception("上传失利")
return QiniuOss.get_url(key), len(content), None
@staticmethod
def get_url(key):
return f"http://oss.pity.fun/{key}"
async def update_file(self, filepath: str, content: bytes, base_path: str = None):
token = self.auth.upload_token(self.bucket, filepath, 3600)
file_name = os.path.basename(filepath)
key = self.get_full_path(filepath, base_path)
ret, info = put_stream(token, key, content, file_name, len(content))
if ret['key'] != key:
raise Exception("更新失利")
async def delete_file(self, filepath: str, base_path: str = None):
key = self.get_full_path(filepath, base_path)
self.bucket_manager.delete(self.bucket, key)
async def list_file(self):
pass
async def download_file(self, filepath, base_path: str = None):
key = self.get_full_path(filepath, base_path)
exists, _ = self.bucket_manager.stat(self.bucket, key)
if exists is None:
raise Exception("文件不存在")
base_url = '%s/%s/%s' % (Config.OSS_URL, self.bucket, filepath)
url = self.auth.private_download_url(base_url, expires=3600)
content, real_name = await self.download_object(key, url)
return content, real_name
async def download_object(self, filepath, url, timeout=15):
async with aiohttp.ClientSession() as session:
async with session.request("GET", url, timeout=timeout, verify_ssl=False) as resp:
if resp.status != 200:
raise Exception("download file failed")
real_filename = filepath.split("/")[-1]
path = rf'./{self.get_random_filename(real_filename)}'
with open(path, 'wb') as f:
data = await resp.content.read()
f.write(data)
return path, real_filename
async def get_file_object(self, filepath):
pass
# if not self.bucket.object_exists(filepath):
# raise Exception(f"oss文件: {filepath}不存在")
# file_object = self.bucket.get_object(filepath)
# return file_object.resp.response.content
这儿直接就贴了代码，咱们略微解说一下。
-
auth
auth是七牛云的认证目标，咱们能够经过它上传/下载文件。
-
bucket_manager
是bucket的办理目标，咱们能够经过bucket去查询目标的信息。
-
download_object
由于七牛云不供给直接下载的功用，而是采用供给一个链接的方式，所以咱们得自己写一个http恳求（异步）去获取url的content数据，并把它写入临时文件，这点和之前比较相似。 -
base_path
之所以需求base_path，是由于咱们的oss分为很多类别，比方用户头像，项目头像，今后还可能会有其他的头像，咱们需求与
测验数据（文件上传用到的测验数据）分开，所以咱们设立了这个根底途径，今后测验数据就都从这个目录获取，而头像则都设置base_path为avatar。说白了就是为了区分数据。 -
get_url
咱们的url都是固定的，自己的域名+存储的途径，所以咱们是能够直接凑集出来的。
-
_convert_to_stream
这个办法比较特别，由于七牛云上传只支撑
文件途径or文件流目标，所以当咱们具有bytes数据的时候，需求转换为BytesIo数据，特此编写了这种办法。
总结一下，七牛云oss和其他oss还是有一些不一样的当地，但毕竟免费，所以就适应一下啦。
修正app/middleware/oss/__init__.py
咱们修正get_oss_client办法:
今后支撑其他oss能够持续再此增加。
测验下页面是否正常
能够看到，下载正常。那其他的博主也跑了跑，由于基本都是调七牛的api，所以也没啥问题。
今天的内容就先介绍到这儿，下一节叙述如何用github action完结主动布置功用，节省大家的时刻！
