pythonrequests上传文件

Z时代
2024-01-10
分类：综合

参考

django 上传文件

临时文件，可以获取名字

内存文件

需求

django 收取到前端上传的文件后转发到微信服务器

requsts 上传文件

https://2.python-requests.org//zh_CN/latest/user/advanced.html#streaming-uploads

流式上传

with open("massive-body") as f:
    requests.post("http://some.url/streamed", data=f)

块编码请求

def gen():
    yield "hi"
    yield "there"
requests.post("http://some.url/chunked", data=gen())

POST 多个分块编码的文件

>>> url = "http://httpbin.org/post"
>>> multiple_files = [
        ("images", ("foo.png", open("foo.png", "rb"), "image/png")),
        ("images", ("bar.png", open("bar.png", "rb"), "image/png"))]
>>> r = requests.post(url, files=multiple_files)
>>> r.text
{
  ...
  "files": {"images": "data:image/png;base64,iVBORw ...."}
  "Content-Type": "multipart/form-data; boundary=3131623adb2043caaeb5538cc7aa0b3a",
  ...
}

requests.models.py

请求编码混入

class RequestEncodingMixin(object):
    @staticmethod
    def _encode_files(files, data):
        """Build the body for a multipart/form-data request.
        Will successfully encode files when passed as a dict or a list of
        tuples. Order is retained if data is a list of tuples but arbitrary
        if parameters are supplied as a dict.
        The tuples may be 2-tuples (filename, fileobj), 3-tuples (filename, fileobj, contentype) # 文件格式
        or 4-tuples (filename, fileobj, contentype, custom_headers).
        """
        if (not files):
            raise ValueError("Files must be provided.")
        elif isinstance(data, basestring):
            raise ValueError("Data must not be a string.")
        new_fields = []
        fields = to_key_val_list(data or {})
        files = to_key_val_list(files or {})
        for field, val in fields:
            if isinstance(val, basestring) or not hasattr(val, "__iter__"):
                val = [val]
            for v in val:
                if v is not None:
                    # Don"t call str() on bytestrings: in Py3 it all goes wrong.
                    if not isinstance(v, bytes):
                        v = str(v)
                    new_fields.append(
                        (field.decode("utf-8") if isinstance(field, bytes) else field,
                         v.encode("utf-8") if isinstance(v, str) else v))
		# 遍历files获取2/3元组/list格式的数据.
        for (k, v) in files:
            # support for explicit filename
            ft = None
            fh = None
            if isinstance(v, (tuple, list)):
                if len(v) == 2:
                    fn, fp = v
                elif len(v) == 3:
                    fn, fp, ft = v
                else:
                    fn, fp, ft, fh = v
            else:
                fn = guess_filename(v) or k
                fp = v
            if isinstance(fp, (str, bytes, bytearray)): # 内存数据
                fdata = fp
            elif hasattr(fp, "read"): # 文件对象, 支持read
                fdata = fp.read()
            elif fp is None:
                continue
            else:
                fdata = fp
            rf = RequestField(name=k, data=fdata, filename=fn, headers=fh)
            rf.make_multipart(content_type=ft)
            new_fields.append(rf)
        body, content_type = encode_multipart_formdata(new_fields)
        return body, content_type

django 中上传文件

settings.py

# 设置上传文件为临时文件，避免使用内存文件 FILE_UPLOAD_HANDLERS = [ "django.core.files.uploadhandler.TemporaryFileUploadHandler", ]

serializers.py 上传文件字段

class MaterialCreateSerializer(serializers.ModelSerializer):
    name = serializers.CharField(
        required=False, max_length=30, help_text=u"标题")
    content = serializers.FileField(
        required=True, help_text=u"材料内容",
        validators=[
            FileExtensionValidator(
                ["png", "jpg", "jpeg", "mp4", "mp3"]
            )
        ]
    )

views.py


class XxxView(generics.GenericAPIView):
    permission_classes = ()
    authentication_classes = ()
    serializer_class = MaterialCreateSerializer
    def post(self, request, *args, **kwargs):
        """
        诗经里景区服务号上传摇一摇功能图片素材
        ---
        parameters:
        - name: object
          pytype: serializers.MaterialCreateSerializer
          paramType: body
        """
        serializer = self.get_serializer(data=request.data)
        if not serializer.is_valid():
            logger.error(
                "WXApiView serializer err:{}".format(serializer.errors))
            return Response(
                serializer.errors, status=status.HTTP_400_BAD_REQUEST)
        data = serializer.validated_data
        url = data.get("url")
        file_data = data.get("image")
        if file_data.closed:
            fp = open(file_data.file.name, "r") # 从不执行
        elif isinstance(file_data, TemporaryUploadedFile):
            fp = file_data.file.file # file对象
	   # 使用requets中的多文件上传, 也可使用但文件
        multiple_files = [
            ("file", ("file", fp, file_data.content_type)),
        ]
        resp = _post(url=url, files=multiple_files)
        if not fp.closed:
            fp.close()
        return resp