七牛云存储

七牛云存储是基于七牛云对象存储服务的云存储解决方案，提供性价比优异的存储能力。它支持多媒体文件处理和图片视频的自动压缩优化，提供简单易用的API接口和快速集成能力，适合中小企业和创业团队的云存储需求。

七牛云存储元素分层结构为Meta（storages.Meta） → Type（storages.QiniuType） → 实例，开发者可通过JitAi的可视化开发工具快捷地创建七牛云存储实例元素。

当然，开发者也可以创建自己的Type元素，或者在自己的App中改写JitAi官方提供的storages.QiniuType元素，以实现自己的封装。

快速开始

创建实例元素

目录结构

推荐目录结构

storages/
└── MyQiniuStorage/
    ├── e.json                          # 元素声明文件
    └── MyQiniuStorage.json            # 七牛云配置文件

e.json文件

元素声明文件

{
  "title": "我的七牛云存储",
  "type": "storages.QiniuType",
  "backendBundleEntry": ".",
  "variables": []
}

业务配置文件

七牛云配置文件

{
  "accessKeyId": "your_access_key_id",
  "accessKeySecret": "your_access_key_secret",
  "bucketName": "your_bucket_name",
  "endPoint": "upload-z2.qiniup.com",
  "scheme": "https"
}

调用示例

使用七牛云存储

# 获取存储实例
storage = app.getElement("storages.MyQiniuStorage")

# 上传文件
with open("example.jpg", "rb") as f:
    file_data = f.read()
    result = storage.uploadByFile("images/example.jpg", file_data, "image/jpeg")
    print(f"文件URL: {result['url']}")

# 获取签名URL
sign_url = storage.getSignUrl("images/example.jpg", "image/jpeg", 3600)
print(f"签名URL: {sign_url}")

# 下载文件
file_content = storage.download("images/example.jpg")

# 删除文件
storage.delete("images/example.jpg")

元素配置

e.json配置

参数名	类型	必填	说明
title	字符串	是	存储实例显示名称
type	字符串	是	固定值：storages.QiniuType
backendBundleEntry	字符串	是	固定值："."
variables	数组	否	元素变量配置，通常为空

业务配置文件配置

参数名	类型	必填	默认值	说明
accessKeyId	字符串	是	-	七牛云Access Key ID
accessKeySecret	字符串	是	-	七牛云Access Key Secret
bucketName	字符串	是	-	七牛云存储空间名称
endPoint	字符串	否	upload-z2.qiniup.com	七牛云上传域名
scheme	字符串	否	http	请求协议，可选http或https

方法

uploadByFile

上传文件到七牛云存储。

参数详解

参数名	类型	对应原生类型	必填	说明
name	Stext	str	是	文件在存储中的路径名称
data	-	bytes	是	文件二进制数据
contentType	Stext	str	否	文件MIME类型，默认为application/octet-stream

返回值

返回包含文件URL的字典，格式：{"url": "文件访问URL"}

使用示例

上传不同类型文件

storage = app.getElement("storages.MyQiniuStorage")

# 上传图片文件
with open("photo.jpg", "rb") as f:
    result = storage.uploadByFile("photos/photo.jpg", f.read(), "image/jpeg")

# 上传PDF文档
with open("document.pdf", "rb") as f:
    result = storage.uploadByFile("docs/document.pdf", f.read(), "application/pdf")

# 上传文本文件
text_data = "Hello, World!".encode('utf-8')
result = storage.uploadByFile("texts/hello.txt", text_data, "text/plain")

getSignUrl

获取文件的访问签名URL，用于安全的文件访问。

参数详解

参数名	类型	对应原生类型	必填	说明
file	Stext	str	是	文件在存储中的路径
contentType	Stext	str	是	文件MIME类型
expires	Numeric	int	否	签名过期时间（秒），默认300秒

返回值

返回文件的签名访问URL字符串。

使用示例

生成签名URL

storage = app.getElement("storages.MyQiniuStorage")

# 生成图片访问链接（默认5分钟过期）
url = storage.getSignUrl("photos/photo.jpg", "image/jpeg")

# 生成文档访问链接（1小时过期）
url = storage.getSignUrl("docs/document.pdf", "application/pdf", 3600)

# 生成视频访问链接（30分钟过期）
url = storage.getSignUrl("videos/demo.mp4", "video/mp4", 1800)

download

下载文件内容。

参数详解

参数名	类型	对应原生类型	必填	说明
name	Stext	str	是	文件在存储中的路径

返回值

返回文件的二进制数据（bytes类型）。

使用示例

下载文件

storage = app.getElement("storages.MyQiniuStorage")

# 下载图片文件
image_data = storage.download("photos/photo.jpg")
with open("downloaded_photo.jpg", "wb") as f:
    f.write(image_data)

# 下载文本文件并读取内容
text_data = storage.download("texts/hello.txt")
text_content = text_data.decode('utf-8')
print(text_content)

getObject

获取文件对象信息和内容。

参数详解

参数名	类型	对应原生类型	必填	说明
name	Stext	str	是	文件在存储中的路径

返回值

返回文件对象，包含文件内容和元数据信息。

使用示例

获取文件对象

storage = app.getElement("storages.MyQiniuStorage")

# 获取文件对象
file_obj = storage.getObject("photos/photo.jpg")
# 处理文件对象（具体结构依赖七牛云SDK）

delete

删除存储中的文件。

参数详解

参数名	类型	对应原生类型	必填	说明
name	Stext	str	是	要删除的文件路径

返回值

返回删除操作结果。

使用示例

删除文件

storage = app.getElement("storages.MyQiniuStorage")

# 删除单个文件
storage.delete("photos/old_photo.jpg")

# 批量删除文件
files_to_delete = ["temp/file1.txt", "temp/file2.txt", "temp/file3.txt"]
for file_path in files_to_delete:
    storage.delete(file_path)

属性

暂无

高级特性

文件类型处理

七牛云存储支持多种文件类型的自动处理和优化，特别适合多媒体文件场景。

多媒体文件处理

storage = app.getElement("storages.MyQiniuStorage")

# 上传并处理图片文件
with open("large_image.jpg", "rb") as f:
    # 图片会自动进行压缩优化
    result = storage.uploadByFile("images/optimized.jpg", f.read(), "image/jpeg")

# 上传视频文件
with open("video.mp4", "rb") as f:
    # 视频支持转码和缩略图生成
    result = storage.uploadByFile("videos/demo.mp4", f.read(), "video/mp4")

批量操作

批量文件管理

storage = app.getElement("storages.MyQiniuStorage")

def batch_upload(file_list):
    """批量上传文件"""
    results = []
    for local_path, remote_path, content_type in file_list:
        with open(local_path, "rb") as f:
            result = storage.uploadByFile(remote_path, f.read(), content_type)
            results.append(result)
    return results

# 批量上传示例
files = [
    ("local/image1.jpg", "images/image1.jpg", "image/jpeg"),
    ("local/doc1.pdf", "documents/doc1.pdf", "application/pdf"),
    ("local/video1.mp4", "videos/video1.mp4", "video/mp4")
]
upload_results = batch_upload(files)

错误处理

异常处理

from jit.errcode import Code

storage = app.getElement("storages.MyQiniuStorage")

try:
    with open("example.jpg", "rb") as f:
        result = storage.uploadByFile("images/example.jpg", f.read(), "image/jpeg")
        print(f"上传成功: {result['url']}")
except Code as e:
    print(f"业务错误: {e}")
except Exception as e:
    print(f"系统错误: {e}")