通过 Pinata API 和 python 固定目录

Pinning directory through Pinata API with python

我正在尝试使用 pinata 的 API 固定整个目录,因为它在他们的 API documentation.

中指定

我也找到了 some python implementation,但这对我来说似乎仍然不起作用。

基本上我从目录中获取所有文件并尝试将它们包含在 post 请求中。

all_files: tp.List[str] = get_all_files(filepath)            
files = {"file": [(file, open(file, "rb")) for file in all_files]}

response: requests.Response = requests.post(url=ipfs_url, files=files, headers=headers)

我得到的错误如下:

ValueError: too many values to unpack (expected 4)

我还尝试了一些硬编码值,但还是不行:

all_files = ['images/0.png', 'images/1.png', 'images/2.png']
files = {"file": [(file, open(file, "rb")) for file in all_files]}

response: requests.Response = requests.post(url=ipfs_url, files=files, headers=headers)

在这种情况下,我得到了一个不同的错误:

TypeError: expected string or bytes-like object

另外,我尝试了这样的尝试:

files = (
    ("file", ("0.png", open('images/0.png', "rb"))),
    ("file", ("1.png", open('images/1.png', "rb")))
)

response: requests.Response = requests.post(url=ipfs_url, files=files, headers=headers)

在这种情况下,Pinata 无法识别我想要的内容并返回以下错误:

{'error': 'More than one file and/or directory was provided for pinning.'}

有人可以给我一些关于我做错了什么的提示吗? 谢谢!

编辑:添加要使用的完整示例:

import os
import requests
import typing as tp

def get_all_files(directory: str) -> tp.List[str]:
    """get a list of absolute paths to every file located in the directory"""
    paths: tp.List[str] = []
    for root, dirs, files_ in os.walk(os.path.abspath(directory)):        
        for file in files_:            
             paths.append(os.path.join(root, file))
    return paths
    
  
print("hello")
# Example 1
all_files: tp.List[str] = get_all_files("images")            
files = {"file": [(file, open(file, "rb")) for file in all_files]}

# Example 2
#all_files = ['images/0.png', 'images/1.png', 'images/2.png']
#files = {"file": [(file, open(file, "rb")) for file in all_files]}

# Example 3
#files = (
#    ("file", ("0.png", open('images/0.png', "rb"))),
#    ("file", ("1.png", open('images/1.png', "rb")))
#)

print ("Files used: \n")
print (files)

headers = {        
    'pinata_api_key': "",
    'pinata_secret_api_key': ""     
}   

ipfs_url = "https://api.pinata.cloud/pinning/pinFileToIPFS"

response: requests.Response = requests.post(url=ipfs_url, files=files, headers=headers)
print(response.json())

好的,我成功上传了整个目录并指定了这样的文件:

files = [
        ('file', ('images/0.png', open('images/0.png', "rb"))),
        ('file', ('images/1.png', open('images/1.png', "rb"))),
    ]

总的来说,要收集所有文件并上传它们,我可以使用以下代码:

all_files: tp.List[str] = get_all_files(directory)            
files = [('file', (file, open(file, "rb"))) for file in all_files]

用这段代码我解决了这个问题。要了解问题,直接从 pinata 上传并查看浏览器进行的调用类型就足够了。

可以看出请求负载有何不同:

  • 缺少元数据,其中必须包含目录名称
  • 对于每个文件,需要以目录/文件名格式定义文件名,但默认请求库将文件名设置为仅文件名,如果打印正文,您会注意到这一点

代码:

from os import walk, path, sep
from requests import Session, Request


def pinata_upload(directory):
    # directory is the abs path of dir
    files = []
    ipfs_url = "https://api.pinata.cloud/pinning/pinFileToIPFS"
    headers = {
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.146 Safari/537.36',
        'pinata_api_key': "your_key",
        'pinata_secret_api_key': "your_secret"
    }
    # directory.split(sep)[-1] is the name of directory
    files.append(('pinataMetadata', (None, '{"name":"' + directory.split(sep)[-1] + '"}')))
    for root, dirs, files_ in walk(path.abspath(directory)):
        for f in files_:
            complete_path = path.join(root, f)
            # sep.join(complete_path.split(sep)[-2:]) create the name of file with the syntax directory/filename
            files.append(('file', (sep.join(complete_path.split(sep)[-2:]), open(complete_path, 'rb'))))
    request = Request(
        'POST',
        ipfs_url,
        headers=headers,
        files=files
    ).prepare()
    response = Session().send(request)
    print(response.request.url)
    print(response.request.headers)
    print(response.request.body)
    print(response.json())


if __name__ == '__main__':
    pinata_upload("/Users/simonesganzerla/Desktop/test")