fastapi-storage

Laravel 风格的 FastAPI 文件存储扩展包，支持多磁盘（disk）管理、统一文件操作接口，以及可扩展自定义驱动。

特性

• Laravel 风格 API：Storage.disk("public").put(...)
• 多驱动支持：local、minio、oss、obs
• 可配置默认磁盘和云磁盘：default / cloud
• 支持自定义驱动：Storage.extend(...)
• 支持自定义驱动配置模型：Storage.register_config_model(...)
• FastAPI 集成：init_storage + get_storage
• 安装后 30 秒可用：fastapi-storage init + init_storage(app, filesystems)
• 同步 + 异步方法（如 put/aput）

安装

基础安装

pip install fastapi-file-storage

按需安装驱动依赖

pip install "fastapi-file-storage[minio]"
pip install "fastapi-file-storage[oss]"
pip install "fastapi-file-storage[obs]"
pip install "fastapi-file-storage[all]"

开发与测试依赖

pip install "fastapi-file-storage[test]"

30 秒上手（最短路径）

pip install fastapi-file-storage
fastapi-storage init
fastapi-storage storage:link

from fastapi import FastAPI
from fastapi_storage.config.filesystems import filesystems
from fastapi_storage.integrations.fastapi import init_storage

app = FastAPI()
init_storage(app, filesystems)

# 现在即可使用 local 磁盘
# storage.disk("local").put("demo.txt", "hello")

如果你希望完全手写配置，可使用下面的方式：

from fastapi_storage import Storage

Storage.configure(
    {
        "default": "local",
        "cloud": "minio",
        "disks": {
            "local": {
                "driver": "local",
                "root": "storage/app",
                "url": "/storage",
            },
            "minio": {
                "driver": "minio",
                "endpoint": "127.0.0.1:9000",
                "access_key": "minioadmin",
                "secret_key": "minioadmin",
                "bucket": "app-bucket",
                "secure": False,
                "domain": "minio.example.com",  # 公开访问域名，不带协议也可
            },
        },
    }
)

Storage.disk("local").put("avatars/a.txt", "hello")
content = Storage.disk("local").get("avatars/a.txt")
print(content.decode())

Laravel 风格 filesystems（安装后即用）

安装后可直接使用包内默认配置：

from fastapi_storage.config.filesystems import filesystems

init_storage 支持直接传 StorageSettings（即 filesystems）、dict 或 None。

初始化脚手架（生成配置文件）

在业务项目根目录执行：

fastapi-storage init

默认生成：

• src/config/filesystems.py
• .env.example

常用命令：

# 已存在时覆盖
fastapi-storage init --force

# 只生成 .env.example
fastapi-storage init --skip-config

# 自定义生成路径
fastapi-storage init --config-path src/config/filesystems.py --env-path .env.example

storage:link（创建公开软链接）

Laravel 风格软链接命令：

fastapi-storage storage:link
# 等价别名
fastapi-storage link

默认行为：

• 创建 public/storage -> storage/app/public
• 若链接已存在则跳过（skipped）
• --force 可覆盖已存在的文件或符号链接
• 若 public/storage 已是目录，返回冲突提示（避免误删目录）

可选参数：

fastapi-storage storage:link --force
fastapi-storage storage:link --link public/storage --target storage/app/public

配置结构

Storage.configure(...) 或 FilesystemManager(settings=...) 接收如下结构：

{
    "default": "local",
    "cloud": "minio",  # 可选，不填时回退 default
    "disks": {
        "local": {"driver": "local", "root": "storage/app", "url": "/storage"},
        "public": {"driver": "local", "root": "storage/app/public", "url": "/storage"},
        "minio": {
            "driver": "minio",
            "endpoint": "127.0.0.1:9000",
            "access_key": "...",
            "secret_key": "...",
            "bucket": "...",
            "secure": False,
            "domain": "minio.example.com",  # 用于生成公开 URL
        },
        "oss": {
            "driver": "oss",
            "endpoint": "oss-cn-hangzhou.aliyuncs.com",
            "access_key": "...",
            "secret_key": "...",
            "bucket": "...",
            "is_cname": False,
            "cdn_domain": "cdn.example.com",  # 用于生成公开 URL
        },
        "obs": {
            "driver": "obs",
            "server": "obs.cn-east-3.myhuaweicloud.com",
            "access_key_id": "...",
            "secret_access_key": "...",
            "bucket": "...",
            "url": "https://obs.example.com/base",  # 用于生成公开 URL
        },
    },
}

环境变量配置（Pydantic Settings）

支持 STORAGE_ 前缀与双下划线嵌套。fastapi-storage init 生成的 .env.example 已包含 local/public/minio/oss/obs 示例，可直接复制修改。

最小可用 local 示例：

STORAGE_DEFAULT=local
STORAGE_DISKS__LOCAL__DRIVER=local
STORAGE_DISKS__LOCAL__ROOT=storage/app
STORAGE_DISKS__LOCAL__URL=/storage

MinIO / OSS / OBS 示例字段（节选）：

说明：公开 URL 相关配置建议显式写完整协议（如 https://...）。若仅填写域名，当前适配器会为云存储公开地址默认补 https://。

STORAGE_DISKS__MINIO__DRIVER=minio
STORAGE_DISKS__MINIO__ENDPOINT=127.0.0.1:9000
STORAGE_DISKS__MINIO__ACCESS_KEY=minioadmin
STORAGE_DISKS__MINIO__SECRET_KEY=minioadmin
STORAGE_DISKS__MINIO__BUCKET=app-bucket
STORAGE_DISKS__MINIO__DOMAIN=minio.example.com

STORAGE_DISKS__OSS__DRIVER=oss
STORAGE_DISKS__OSS__ENDPOINT=oss-cn-hangzhou.aliyuncs.com
STORAGE_DISKS__OSS__ACCESS_KEY=your-access-key
STORAGE_DISKS__OSS__SECRET_KEY=your-secret-key
STORAGE_DISKS__OSS__BUCKET=your-bucket
STORAGE_DISKS__OSS__CDN_DOMAIN=cdn.example.com

STORAGE_DISKS__OBS__DRIVER=obs
STORAGE_DISKS__OBS__SERVER=obs.cn-east-3.myhuaweicloud.com
STORAGE_DISKS__OBS__ACCESS_KEY_ID=your-access-key-id
STORAGE_DISKS__OBS__SECRET_ACCESS_KEY=your-secret-access-key
STORAGE_DISKS__OBS__BUCKET=your-bucket
STORAGE_DISKS__OBS__URL=https://obs.example.com/base

API 概览

门面 API

• Storage.configure(settings | manager)
• Storage.disk(name=None)
• Storage.drive(name=None)
• Storage.cloud()
• Storage.extend(driver, factory)
• Storage.register_config_model(driver, model)
• Storage.build(config)（构建临时磁盘）

适配器通用方法

• put(path, content) / aput(...)
• get(path) / aget(...)
• exists(path) / aexists(...)
• delete(path | [paths]) / adelete(...)
• url(path)
• internal_url(path) / ainternal_url(...)
• size(path) / asize(...)
• copy(src, dst) / acopy(...)
• move(src, dst) / amove(...)
• presigned_upload_url(path, ...) / apresigned_upload_url(...)（MinIO）
• presigned_upload_urls(paths, ...) / apresigned_upload_urls(...)（MinIO 批量生成）
• initiate_multipart_upload(path, ...) / ainitiate_multipart_upload(...)（MinIO）
• complete_multipart_upload(path, ...) / acomplete_multipart_upload(...)（MinIO）
• abort_multipart_upload(path, ...) / aabort_multipart_upload(...)（MinIO）

MinIO 预签名上传

MinIO 适配器支持生成客户端直传用的 PUT 预签名 URL。配置了 domain 时，签名地址会使用 domain 生成；未配置时回退到内部 endpoint。

from fastapi_storage import Storage

disk = Storage.disk("minio")

upload_id = disk.initiate_multipart_upload(
    "uploads/video/source.mp4",
    content_type="video/mp4",
)

part_urls = disk.presigned_upload_urls(
    {
        "1": "uploads/video/source.mp4",
        "2": "uploads/video/source.mp4",
    },
    upload_id=upload_id,
    expires=3600,
)

result = disk.complete_multipart_upload(
    "uploads/video/source.mp4",
    upload_id=upload_id,
    parts=[
        {"partNumber": 1, "etag": "part-1-etag"},
        {"partNumber": 2, "etag": "part-2-etag"},
    ],
)

# 上传失败或取消时：
disk.abort_multipart_upload("uploads/video/source.mp4", upload_id=upload_id)

FastAPI 集成

from fastapi import Depends, FastAPI
from fastapi_storage.config.filesystems import filesystems
from fastapi_storage.integrations.fastapi import get_storage, init_storage
from fastapi_storage.manager import FilesystemManager

app = FastAPI()

# 支持 StorageSettings / dict / None
init_storage(app, filesystems)

@app.post("/upload")
def upload(storage: FilesystemManager = Depends(get_storage)):
    storage.disk("local").put("demo.txt", "hello")
    return {"ok": True}

FastAPI 高并发推荐用法（async）

在高并发场景下，建议在 async def 路由中使用异步方法（aput/aget/...），避免阻塞事件循环：

from fastapi import Depends, FastAPI
from fastapi_storage.integrations.fastapi import get_storage, init_storage
from fastapi_storage.manager import FilesystemManager

app = FastAPI()
init_storage(app)


@app.post("/upload-async")
async def upload_async(storage: FilesystemManager = Depends(get_storage)):
    disk = storage.disk("local")
    await disk.aput("demo.txt", "hello", content_type="text/plain")
    data = await disk.aget("demo.txt")
    exists = await disk.aexists("demo.txt")
    return {
        "ok": True,
        "exists": exists,
        "content": data.decode("utf-8"),
    }

说明：

• 同步方法：put/get/exists/delete/...
• 异步方法：aput/aget/aexists/adelete/...
• 异步方法适合 FastAPI 并发接口；同步方法适合同步脚本或后台任务。

自定义驱动示例

from pydantic import BaseModel
from fastapi_storage import Storage


class MemoryDiskConfig(BaseModel):
    driver: str = "memory"


class MemoryAdapter:
    def __init__(self, config: MemoryDiskConfig):
        self.config = config
        self.data = {}

    def put(self, path, content, **kwargs):
        self.data[path] = content if isinstance(content, bytes) else str(content).encode()
        return True

    def get(self, path):
        return self.data[path]

    def exists(self, path):
        return path in self.data

    def delete(self, paths):
        targets = [paths] if isinstance(paths, str) else paths
        for p in targets:
            self.data.pop(p, None)
        return True

    def url(self, path):
        return f"memory://{path}"

    def size(self, path):
        return len(self.data[path])

    def copy(self, src, dst):
        self.data[dst] = self.data[src]
        return True

    def move(self, src, dst):
        self.data[dst] = self.data[src]
        del self.data[src]
        return True


Storage.configure(
    {
        "default": "memory_disk",
        "disks": {
            "memory_disk": {"driver": "memory"},
        },
    }
)

Storage.register_config_model("memory", MemoryDiskConfig)
Storage.extend("memory", lambda config: MemoryAdapter(config))

Storage.disk("memory_disk").put("a.txt", "hello")
print(Storage.disk("memory_disk").get("a.txt"))

本地开发

pip install -e ".[test]"
python3 -m pytest tests

更新历史

0.1.5

• 新增 MinIO multipart initiate / complete / abort 封装
• multipart complete 支持 partNumber / part_number / tuple part 入参
• 补充 multipart 生命周期同步 / 异步单测

0.1.4

• 新增 MinIO 预签名上传 URL 生成方法，支持单个对象、批量对象和 multipart part 上传
• 配置 domain 时，MinIO 预签名上传地址使用公开域名生成，而不是内部 endpoint
• 补充预签名上传的同步 / 异步接口和单测覆盖

0.1.3

• 修复 MinIO / OSS / OBS 在仅配置域名（不带 scheme）时生成的公开 URL 不完整问题
• 统一云存储 url() 空路径行为（返回空串）
• 修复 url(None) / internal_url(None) 触发 AttributeError 的问题

0.1.2

• 新增 internal_url(path) / ainternal_url(path) 接口
• 新增 fput(path, file) / afput(path, file) 接口
• 第三方存储驱动在无原生支持时，internal_url 使用适配层基于当前 disk 配置的兼容实现
• 第三方存储驱动在无原生支持时，fput 使用“读取本地文件后再调用 put/aput”的兼容实现
• 补充 contract / async / integration 测试覆盖

0.1.1

• 重命名 PyPI 包为 fastapi-file-storage
• 补充 .gitignore 与发布相关整理

发布流程

完整发布步骤见 RELEASE.md。