使用urllib3模块而不是python中的请求

问题描述 投票:1回答:2

此代码在python 2.X版本中正常工作。我正在尝试在python版本3中使用类似的代码。问题是我不想使用请求模块。我需要使用“ urllib3”使其工作。

import requests
import urllib

event = {'url':'http://google.com', 'email':'[email protected]', 'title':'test'}

url = event['url']
if event['email']:
    email=event['email']

if event['title']:
    title=event['title']

url1 = urllib.parse.unquote(url)
myfile=urllib.request.urlopen(url1)

requests.post("https://api.mailgun.net/v3/xxx.mailgun.org/messages",
                    auth=("api", "key-xxx"),
                    files=[("attachment", myfile)
                           ],
                    data={"from": "Excited User <[email protected]>",
                          "to": email,
                          "subject": title,
                          "text": "Testing some awesomness with attachments!",
                          "html": myfile})

[我在尝试运行此代码时遇到TypeError:

import urllib3

event = {'url':'http://oksoft.blogspot.com', 'email':'[email protected]', 'title':'test'}

url = event['url']
if event['email']:
    email=event['email']

if event['title']:
    title=event['title']

url1 = urllib.parse.unquote(url)
myfile=urllib.request.urlopen(url1)

http = urllib3.PoolManager()
url = "https://api.mailgun.net/v3/xxx.mailgun.org/messages"

params={"from": "Excited User <[email protected]>",
                          "to": email,
                          "subject": title,
                          "text": "Testing some awesomness with attachments!",
                          "html": myfile} 

http.request(
    "POST", url, headers={"Content-Type": "application/json", "api":"key-xxx"}, body= params
)
python python-3.x aws-lambda python-requests urllib3
2个回答
0
投票

它返回TypeError: can't concat str to bytes,因为您的myfilehttp.client.HTTPResponse类型,而不是str类型-需要将其传递给请求正文。

因此必须首先将响应转换为字符串,然后对正文params进行json序列化。


完整代码:

import urllib3
import urllib.request
import urllib.parse
import json

event = {'url': 'http://oksoft.blogspot.com',
         'email': '[email protected]', 'title': 'test'}

url = event['url']
if event['email']:
    email = event['email']

if event['title']:
    title = event['title']

url1 = urllib.parse.unquote(url)
myfile = urllib.request.urlopen(url1)

myfile_content = myfile.read()\
    .decode(myfile.headers
            .get_content_charset(failobj='utf-8'))

http = urllib3.PoolManager()
url = "https://api.mailgun.net/v3/xxx.mailgun.org/messages"

params = {"from": "Excited User <[email protected]>",
          "to": email,
          "subject": title,
          "text": "Testing some awesomness with attachments!",
          "html": myfile_content}

### EDIT

auth_headers = urllib3.util.make_headers(
    basic_auth='api:xxx')

r = http.request(
    "POST",
    url,
    headers=auth_headers,
    fields=params
)

print(r.status, r.data)

0
投票

此代码使用urllib.reqeust模块。与inline html内容相对,实际创建文件attachment的过程比为文件内容设置html参数要复杂得多。

import urllib.request
import urllib.error
import urllib.parse

import io
import mimetypes
import uuid

class MultiPartForm:
    """Accumulate the data to be used when posting a form."""

    def __init__(self):
        self.form_fields = []
        self.files = []
        # Use a large random byte string to separate
        # parts of the MIME data.
        self.boundary = uuid.uuid4().hex.encode('utf-8')
        return

    def get_content_type(self):
        return 'multipart/form-data; boundary={}'.format(
            self.boundary.decode('utf-8'))

    def add_field(self, name, value):
        """Add a simple field to the form data."""
        self.form_fields.append((name, value))

    def add_file(self, fieldname, filename, fileHandle,
                 mimetype=None):
        """Add a file to be uploaded."""
        body = fileHandle.read()
        if mimetype is None:
            mimetype = (
                mimetypes.guess_type(filename)[0] or
                'application/octet-stream'
            )
        self.files.append((fieldname, filename, mimetype, body))
        return

    @staticmethod
    def _form_data(name):
        return ('Content-Disposition: form-data; '
                'name="{}"\r\n').format(name).encode('utf-8')

    @staticmethod
    def _attached_file(name, filename):
        return ('Content-Disposition: file; '
                'name="{}"; filename="{}"\r\n').format(
                    name, filename).encode('utf-8')

    @staticmethod
    def _content_type(ct):
        return 'Content-Type: {}\r\n'.format(ct).encode('utf-8')

    def __bytes__(self):
        """Return a byte-string representing the form data,
        including attached files.
        """
        buffer = io.BytesIO()
        boundary = b'--' + self.boundary + b'\r\n'

        # Add the form fields
        for name, value in self.form_fields:
            buffer.write(boundary)
            buffer.write(self._form_data(name))
            buffer.write(b'\r\n')
            buffer.write(value.encode('utf-8'))
            buffer.write(b'\r\n')

        # Add the files to upload
        for f_name, filename, f_content_type, body in self.files:
            buffer.write(boundary)
            buffer.write(self._attached_file(f_name, filename))
            buffer.write(self._content_type(f_content_type))
            buffer.write(b'\r\n')
            buffer.write(body)
            buffer.write(b'\r\n')

        buffer.write(b'--' + self.boundary + b'--\r\n')
        return buffer.getvalue()


event = {'url':'http://oksoft.blogspot.com', 'email':'[email protected]', 'title':'test'}

url = event['url']
if event['email']:
    email=event['email']

if event['title']:
    title=event['title']

form = MultiPartForm()
form.add_field("from", "Excited User <[email protected]>")
form.add_field("to", email)
form.add_field("subject", title)
form.add_field("text", "Testing some awesomness with attachments!")
with urllib.request.urlopen(url) as f:
    form.add_file("attachment", "test.html", f, "text/html")

url = "https://api.mailgun.net/v3/xxx.mailgun.org/messages"

# create basic authorization opener
auth_handler = urllib.request.HTTPBasicAuthHandler()
auth_handler.add_password(realm='MG API',
                          uri=url,
                          user='api',
                          passwd='xxx-key')
opener = urllib.request.build_opener(auth_handler)

data = bytes(form)
req = urllib.request.Request(url, data=data)
req.add_header('Content-type', form.get_content_type())
req.add_header('Content-length', len(data))
with opener.open(req) as f:
     print(f.read().decode('utf-8'))
© www.soinside.com 2019 - 2024. All rights reserved.