This is the 29th day of my participation in the Gwen Challenge in November. Check out the details: The last Gwen Challenge in 2021
Hello ~ I’m Milo! I am building an open source interface testing platform from 0 to 1, and I am also writing a complete tutorial corresponding to it. I hope you can support it. Welcome to pay attention to my public number test development pit goods, get the latest article tutorial!
review
In the previous section, we wrote the online test plan functionality and tweaked the report page slightly. That’s actually a lot of holes in our previous content.
For example, HTTP requests only support JSON and form, but do not support file upload requests. Even some CRUD functions are not perfect.
But never mind, my idea is to create first, and then perfect. Of course, it is not blind creation, and it is necessary to predict the future in advance.
Why use OSS
Pity is intended for two purposes. The first is some static resource images. Like project images, user avatars.
The other thing is that when we test the file upload interface, we need to test the file. How do you get these documents and manage them? All with OSS.
Currently, taking oss that I am familiar with as an example, WE intend to support the following oss:
-
Ali Cloud OSS ➤
-
Tencent Cloud COS (no account) 💊
-
Seven Niuyun (free, I can show demo)
-
Gitee (free)
Among them, Ali cloud and Tencent cloud are paid products, and bloggers can’t afford them. Therefore, it may be necessary to have a corresponding test account. Here I would like to thank Mini-Right for giving me a test account of Aliyun, which ensures the normal operation of CRUD.
If you need me to implement other OSS clients, please bring the corresponding test account private message to me.
The data from the test site will probably be gitee or Qiuniuyun.
instructions
As for file management, due to time constraints, I will not associate a table with OSS data for the time being. The main purpose is to save time.
Table association can be used as a phase II project.
Write oss base classes
The new app. Middleware. Oss. Oss_file. Py
Since we support multiple OSS clients, we pack an abstract class (like go interface) for each OSS client to implement.
from abc import ABC, abstractmethod
from typing import ByteString
class OssFile(ABC) :
@abstractmethod
def create_file(self, filepath: str, content: ByteString) :
pass
@abstractmethod
def update_file(self, filepath: str, content: ByteString) :
pass
@abstractmethod
def delete_file(self, filepath: str) :
pass
@abstractmethod
def list_file(self) :
pass
@abstractmethod
def download_file(self, filepath) :
pass
Copy the codeAbstract classes have no concrete implementation, only method definitions, which are intended to limit the methods of subclasses, that is, methods defined in the base class must be implemented.
To sum up, there must be 5 ways to add, delete, change, check and download.
Write the AliyunOss implementation
The new app. Middleware. Oss. Aliyun. Py
from typing import ByteString
import oss2
from app.middleware.oss.files import OssFile
class AliyunOss(OssFile) :
def __init__(self, access_key_id: str, access_key_secret: str, endpoint: str, bucket: str) :
auth = oss2.Auth(access_key_id=access_key_id,
access_key_secret=access_key_secret)
# auth = oss2.AnonymousAuth()
self.bucket = oss2.Bucket(auth, endpoint, bucket)
def create_file(self, filepath: str, content: ByteString) :
self.bucket.put_object(filepath, content)
def update_file(self, filepath: str, content: ByteString) :
self.bucket.put_object(filepath, content)
def delete_file(self, filepath: str) :
self.bucket.delete_object(filepath)
def list_file(self) :
ans = []
for obj in oss2.ObjectIteratorV2(self.bucket):
ans.append(dict(key=obj.key, last_modified=obj.last_modified,
size=obj.size, owner=obj.owner))
return ans
def download_file(self, filepath) :
if not self.bucket.object_exists(filepath):
raise Exception(F "oss file:{filepath}There is no")
return self.bucket.get_object(filepath)
Copy the codeAliyunOss inherits OssFile. The constructor gets the identity information of Alicloud and verifies it. Finally, read the bucket. This bucket is the area where your files are stored. Let’s call it disk F.
The other method is quite simple, basically calling the corresponding API to do the CRUD operation. Oss does not have a folder concept, but uses a unified path to store file addresses, for example:
woody/github.txt
This file path can be interpreted as github. TXT in the woody directory.
Write a method to get the client
app.middleware.oss._\_init__.py
from app.core.configuration import SystemConfiguration
from app.middleware.oss.aliyun import AliyunOss
from app.middleware.oss.files import OssFile
class OssClient(object) :
_client = None
@classmethod
def get_oss_client(cls) -> OssFile:
""" Retrieving the OSS client through the OSS configuration :return: """
if OssClient._client is None:
cfg = SystemConfiguration.get_config()
oss_config = cfg.get("oss")
access_key_id = oss_config.get("access_key_id")
access_key_secret = oss_config.get("access_key_secret")
bucket = oss_config.get("bucket")
endpoint = oss_config.get("endpoint")
if oss_config is None:
raise Exception("The server is not configured with OSS information, please add it in configuration.json")
if oss_config.get("type").lower() == "aliyun":
return AliyunOss(access_key_id, access_key_secret, endpoint, bucket)
raise Exception("Unsupported OSS types")
return OssClient._client
Copy the codeWe configure the OSS information in configuration.json and then get the client from OssClient each time.
Write the back-end interface
from fastapi import APIRouter, File, UploadFile
from app.handler.fatcory import PityResponse
from app.middleware.oss import OssClient
router = APIRouter(prefix="/oss")
@router.post("/upload")
async def create_oss_file(filepath: str, file: UploadFile = File(.)) :
try:
file_content = await file.read()
Obtain the OSS client
client = OssClient.get_oss_client()
client.create_file(filepath, file_content)
return PityResponse.success()
except Exception as e:
return PityResponse.failed(F "Upload failed:{e}")
@router.get("/list")
async def list_oss_file() :
try:
client = OssClient.get_oss_client()
files = client.list_file()
return PityResponse.success(files)
except Exception as e:
return PityResponse.failed(F "fetch failed:{e}")
@router.get("/delete")
async def delete_oss_file(filepath: str) :
try:
client = OssClient.get_oss_client()
client.delete_file(filepath)
return PityResponse.success()
except Exception as e:
return PityResponse.failed(F "delete failed:{e}")
@router.post("/update")
async def update_oss_file(filepath: str, file: UploadFile = File(.)) :
"""
更新oss文件，路径不能变化
:param filepath:
:param file:
:return:
"""
try:
client = OssClient.get_oss_client()
file_content = await file.read()
client.update_file(filepath, file_content)
return PityResponse.success()
except Exception as e:
return PityResponse.failed(F "delete failed:{e}")
Copy the codeThe method is very simple, the File path is in the URL parameter, fastAPI File and UploadFile can be used to obtain the uploaded File.
Note that you must first install the Python-Multipart library to accompany file upload
Test the
Oss file information was successfully read, but the interface was downloaded
I seem to have forgotten about downloading files, so refer to my previous article about downloading files from FsatApi.
FastApi downloads files
Or go directly to Github to view the source ~
That’s it for today, the next section on oss pages.