fastapi+minio:v6

这应该是最终版本，加入了缓存机制

config/config.py

@@ -6,13 +6,13 @@
 """
 import json
 from minio import Minio
-
+from cacheout import Cache
 
 # 从配置文件读取设置
 class MinioOperate:
 
     def __init__(self):
-        with open(r"config/config.json", "r") as f:
+        with open(r"D:\pythonProject\django\fastapi_01\config\config.json", "r") as f:
             self.__config = json.load(f)
         self.minio_client = None
 
@@ -30,4 +30,16 @@ class MinioOperate:
             else:
                 print(f"Bucket {bucket_name} already exists")
 
+class SetCache:
+    def __init__(self,maxsize,ttl):
+        self.cache = Cache(maxsize=maxsize, ttl=ttl)
+
+    def get(self,uid):
+        self.data = self.cache.get(uid)
+        return self.data
+
+    def add(self,uid,data):
+        self.cache.add(uid,data)
+
+
 

pydantic_test.py

@@ -1,62 +0,0 @@
-"""
-    encoding: UTF-8
-    @author:clx
-    @file:pydantic_test.py
-    @time:2023/5/29 22:55
-"""
-import random
-
-import fastapi
-
-from starlette.responses import FileResponse
-# from fastapi.responses import FileResponse
-# from datetime import datetime
-# from typing import Optional, List
-#
-# from pydantic import BaseModel
-# from fastapi import FastAPI
-#
-# app = FastAPI()
-#
-# class User(BaseModel):
-#     id:int
-#     name:str = "Jone"
-#     signup_ts = Optional[datetime]
-#     friends:List[int]
-#
-# dic = {
-#     "id":1,
-#     "signup_ts":"2022-11-21",
-#     "friends":[1,2,"3"]
-# }
-# user = User(**dic)
-# print(user.id)
-
-import os
-from pathlib import Path
-
-def fun():
-    f = os.path.join(r"./", "name")
-
-    print(f)
-    print(Path(f).name)
-
-    fr = FileResponse(
-        path=f,
-        filename=Path(f).name
-    )
-
-# fun()
-
-# a = "dsadas.txt"
-# file_ext = os.path.splitext(a)
-# front,ext = file_ext
-# print(front,ext)
-
-# print("".join(random.sample("qwertyuiopasdfghjklzxcvbnm1234567890",5)))
-
-import time
-import re
-a = str(time.time())
-b = re.sub(r"\.","",a)
-print(b)

routers/files.py

@@ -6,12 +6,13 @@ import datetime
 from io import BytesIO
 from fastapi import Depends, FastAPI, HTTPException, status, UploadFile, APIRouter, File
 from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
-from starlette.responses import StreamingResponse
+from starlette.responses import StreamingResponse, FileResponse
 from config import config
 from models.model import *
 from utils.jwt import authenticate_user, ACCESS_TOKEN_EXPIRE_MINUTES, create_access_token, get_current_active_user, \
     fake_users_db
 
+
 # 创建minio对象
 minio_class = config.MinioOperate()
 # 连接minio
@@ -19,8 +20,10 @@ minio_client = minio_class.link_minio()
 # 创建bucket
 minio_class.create_bucket(["file", "image"])
 
-router = APIRouter()
+# 初始化缓存
+cache = config.SetCache(maxsize=128,ttl=10)
 
+router = APIRouter()
 
 @router.post("/token", response_model=Token)
 async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
@@ -43,7 +46,7 @@ async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(
 # minio ?上传文件时在重复时，需要修改文件名
 @router.post("/file")
 async def create_file(file: UploadFile = File(...)):
-    timestamp = str(time.time()).ljust(18,"0")
+    timestamp = str(time.time()).ljust(18, "0")
     uid = re.sub(r"\.", "", timestamp)
     front, ext = os.path.splitext(file.filename)
 
@@ -65,24 +68,33 @@ async def create_file(file: UploadFile = File(...)):
         return {"status": 400, "data": [], "msg": "Post Failed!"}
 
 
-# 图片预览
+# 图片预览【把后缀jpg去掉也可以变成下载】
 # response = StreamingResponse(file_content,media_type='image/jpg)
-
 # 图片下载功能
+# response = StreamingResponse(file_content,
+#                              media_type='application/octet-stream',
+#                              headers={"Content-Disposition": f"attachment; filename={uid}"})
+
 @router.get("/file/{uid}")
 async def download_file(uid: str):
     try:
         timestamp, ext = os.path.splitext(uid)
-        timestamp = float(str(float(timestamp) / 10000000).ljust(18,"0"))
+        timestamp = float(str(float(timestamp) / 10000000).ljust(18, "0"))
         object_path = str(time.localtime(timestamp).tm_year) + "-" + str(time.localtime(timestamp).tm_mon).rjust(2,
                                                                                                                  "0") + "-" \
                       + str(time.localtime(timestamp).tm_mday).rjust(2, "0") + "/{}".format(uid)
         file_obj = minio_client.get_object("image", object_path)
+
+        if not cache.get(uid):
+            # 添加缓存
+            print("第一次获取，添加到缓存")
+            cache.add(uid, file_obj.read())
+        else:
+            print("从缓存中找到uid，获取缓存")
+            file_bytes = cache.get(uid)
+            return StreamingResponse(BytesIO(file_bytes), media_type="image/{}".format(ext[1:]))
         file_content = BytesIO(file_obj.read())
-        # response = StreamingResponse(file_content,
-        #                              media_type='application/octet-stream',
-        #                              headers={"Content-Disposition": f"attachment; filename={uid}"})
-        response = StreamingResponse(file_content, media_type='image/png')
+        response = StreamingResponse(file_content, media_type='image/{}'.format(ext[1:]))
     except Exception as e:
         return {"status": 400, "data": [], "msg": "Get Failed!"}
     return response
@@ -93,7 +105,7 @@ async def download_file(uid: str):
 async def delete_file(uid: str):
     try:
         timestamp, ext = os.path.splitext(uid)
-        timestamp = float(str(float(timestamp) / 10000000).ljust(18,"0"))
+        timestamp = float(str(float(timestamp) / 10000000).ljust(18, "0"))
         object_path = str(time.localtime(timestamp).tm_year) + "-" + str(time.localtime(timestamp).tm_mon).rjust(2,
                                                                                                                  "0") + "-" \
                       + str(time.localtime(timestamp).tm_mday).rjust(2, "0") + "/{}".format(uid)

test_unit.py

@@ -1,25 +0,0 @@
-"""
-    encoding: UTF-8
-    @author:clx
-    @file:test_unit.py
-    @time:2023/5/30 20:06
-"""
-import uvicorn
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-import pytest
-
-app = FastAPI()
-
-client = TestClient(app)
-
-
-def test_create_file():
-    response = client.get("/item/res")
-    assert response.status_code == 200
-    assert response.json() == [
-    {"id": 1, "name": "jack"},
-    {"id": 2, "name": "mark"},
-    {"id": 3, "name": "tom"},
-    {"id": 4, "name": "sery"}
-]