创建`OSS_minio.py`并定义`upload_stream`和`download_stream` ```py class OSS: # Create a client with the MinIO server playground, its access key # and secret key. def __init__(self, host, username, password, bucket): self.host = host self.client = Minio( host, access_key=username, secret_key=password, secure=False ) self.bucket = bucket def upload_stream(self,remote_path,data): try: self.client.put_object(self.bucket,remote_path,data,-1,part_size=5*1024*1024) print( "file is successfully uploaded as \n object %s to bucket %s." % ( remote_path, self.bucket) ) address = 'http://'+self.host+'/'+self.bucket+'/'+remote_path print(address) return address except S3Error as exc: print("error occurred.", exc) def download_stream(self, remote_path): try: response = self.client.get_object(self.bucket, remote_path) return response except S3Error as exc: print("error occurred.", exc) def init(): host = "" username = "" password = "" bucket = "" return OSS(host, username, password, bucket) ``` 创建文件`templates/index.html` ```html

Transfery

``` 编写flask ```py import OSS_minio import flask client=OSS_minio.init() app = flask.Flask(__name__) @app.route('/', methods=['POST', 'GET']) def index(): if flask.request.method == 'POST': f = flask.request.files.get('file') if f: print('uploading ...') client.upload_stream(f.filename,f) #流式上传 return flask.redirect(flask.url_for('index')) else: return flask.redirect(flask.url_for('download_stream')) return flask.render_template('index.html') @app.route('/download') def download_stream(): file_name='app.py' print('downloading ...') file_stream=client.download_stream(file_name) #流式下载 return flask.send_file(file_stream,download_name=file_name,as_attachment=True) ```