用twisted 发送请求

用twisted 发送请求

# coding: utf-8# @Time : 2022-05-18 13:01# @Author : AngDHfrom twisted.internet import reactorfrom twisted.web.client import Agentfrom twisted.web.import Headers# 相当于客户端角色agent = Agent(reactor)defered = agent.request( b'GET', b' Headers({"User-Agent": ["python"]}), None)def successCallback(response): print('Response received:', response.code)defered.addCallback(successCallback)def errorCallback(error): print("error:", str(error))defered.addErrback(errorCallback)def callbackShutdown(ignored): reactor.stop()defered.addBoth(callbackShutdown)reactor.run()

# coding: utf-8# @Time : 2022-05-18 13:01# @Author : AngDHfrom twisted.internet import reactorfrom twisted.internet.defer import succeedfrom twisted.web.client import Agentfrom twisted.web.import Headersfrom twisted.web.iweb import IBodyProducerfrom zope.interface import implementer@implementer(IBodyProducer)class BytesProducer(object): def __init__(self, body): self.body = body self.length = len(body) def startProducing(self, consumer): consumer.write(self.body) return succeed(None) def pauseProducting(self): pass def stopProducing(self): pass# 相当于客户端角色agent = Agent(reactor)defered = agent.request( b'POST', b' Headers({"User-Agent": ["python"]}), BytesProducer('请求体'.encode()))def successCallback(response): print('Response received:', response.code)defered.addCallback(successCallback)def errorCallback(error): print("error:", str(error))defered.addErrback(errorCallback)def callbackShutdown(ignored): reactor.stop()defered.addBoth(callbackShutdown)reactor.run()

# coding: utf-8# @Time : 2022-05-18 13:01# @Author : AngDHfrom twisted.internet import reactorfrom twisted.internet.defer import DeferredListfrom twisted.web.client import Agentfrom twisted.web.import Headers# 相当于客户端角色agent = Agent(reactor)def successCallback(response): print('Response received:', response.code)def errorCallback(error): print("error:", str(error))def callbackShutdown(ignored): reactor.stop()_ = []for url in [b" b" b" b" b" b" ]: d = agent.request( b'GET', url, Headers({"User-Agent": ["python"]}), None) d.addCallback(successCallback) d.addErrback(errorCallback) _.append(d)# 统一管理 多个defered对象dl = DeferredList(_)dl.addBoth(callbackShutdown)reactor.run()

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。