如何通过继承对 python 中的请求进行子类化
How to subclass requests in python through inheritance
我想专门化/子class请求包以添加一些具有自定义功能的方法。
我试过这样做:
# concrete_requests.py
import requests
class concreteRequests(requests):
def __init__(self):
super(concreteRequests, self).__init__()
self.session()
def login(self):
payload = {'user': 'foo', 'pass': 'bar'}
self.get('loginUrl', headers=header, data=payload)
# more login stuff...
# my_class.py
class MyClass:
def __init__():
self.requests = concreteRequests()
self.requests.login()
这样我仍然可以受益于 self.requests
成员 + 我的具体实施。所以我可以这样做:self.requests.get(...)
或 print(self.requests.post(...).status_code)
等等。
我猜这行 super(concreteRequests, self).__init__()
可能毫无用处,因为 requests 中没有任何 class 声明,只是导入...
那么,requests包可以subclassed/specialized通过继承吗?
requests
是 python 模块而不是 class。你只能子class classes.
所以基本上你应该在你自己的内部利用它的 methods/functions
自定义 class.
import requests
class MyRequests:
def __init__(self, username, passwd):
self.username = username
self.passwd = passwd
def get(self, *args, **kwargs):
# do your thing
resp = requests.get(...)
# do more processing
我上面写的只是一个让你开始的例子。
我想给你一个完整的例子:
# ----example.py----
from locust import HttpUser, task, between
import requests
# Create fake api endpoint POST https://mocki.io/fake-json-api
class MyClient():
def __init__(self, host):
self.session = requests.session()
self.host = host
def create_user(self, *args, **kwargs):
return self.session.post('{}/v1/9b021e1a'.format(self.host), json={"name":"something"})
class QuickstartUser(HttpUser):
wait_time = between(1, 2)
@task
def api(self):
my_client = MyClient('https://api.mocki.io')
my_client.session = self.client
print(my_client.create_user().status_code)
print('Executing client normally')
client = MyClient('https://api.mocki.io')
print(client.create_user().status_code)
您可以执行客户端代码:
python example.py
你可以用locust执行客户端:
locust --host https://api.mocki.io --users 5 --spawn-rate 1 --locustfile example.py
在两次执行中,您使用的是同一个客户端。
一个好的方法是从请求中继承 Session
object。一个基本的例子:
from requests import Session
class MyClient(Session):
"""Specialized client that inherits the requests api."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def get_google(self):
return self.get("http://google.com")
MyClient
免费包含 Session (request
) api,以及您想要添加的任何其他内容。
A real-world 示例:假设客户端需要在运行时指定的身份验证 header(在这种情况下,身份验证需要当前时间戳)。这是一个子类 Session
和子类 AuthBase
的示例客户端,它实现了这一点(此代码需要为 api_key、secret_key、密码短语设置值):
import json, hmac, hashlib, time, requests, base64
from requests.auth import AuthBase
from requests import Session
class MyClient(Session):
"""Client with specialized auth required by api."""
def __init__(self, api_key, secret_key, passphrase, *args, **kwargs):
# allow passing args to `Session.__init__`
super().__init__(*args, **kwargs)
# `self.auth` callable that creates timestamp when request is made
self.auth = MyAuth(api_key, secret_key, passphrase)
class MyAuth(AuthBase):
"""Auth includes current timestamp when called.
https://docs.python-requests.org/en/master/user/advanced/#custom-authentication
"""
def __init__(self, api_key, secret_key, passphrase):
self.api_key = api_key
self.secret_key = secret_key
self.passphrase = passphrase
def __call__(self, request):
timestamp = str(time.time())
message = timestamp + request.method + request.path_url + (request.body or "")
message = message.encode("utf-8")
hmac_key = base64.b64decode(self.secret_key)
signature = hmac.new(hmac_key, message, hashlib.sha256)
signature_b64 = base64.b64encode(signature.digest())
request.headers.update(
{
"ACCESS-SIGN": signature_b64,
"ACCESS-TIMESTAMP": timestamp,
"ACCESS-KEY": self.api_key,
"ACCESS-PASSPHRASE": self.passphrase,
"Content-Type": "application/json",
}
)
return request
我想专门化/子class请求包以添加一些具有自定义功能的方法。
我试过这样做:
# concrete_requests.py
import requests
class concreteRequests(requests):
def __init__(self):
super(concreteRequests, self).__init__()
self.session()
def login(self):
payload = {'user': 'foo', 'pass': 'bar'}
self.get('loginUrl', headers=header, data=payload)
# more login stuff...
# my_class.py
class MyClass:
def __init__():
self.requests = concreteRequests()
self.requests.login()
这样我仍然可以受益于 self.requests
成员 + 我的具体实施。所以我可以这样做:self.requests.get(...)
或 print(self.requests.post(...).status_code)
等等。
我猜这行 super(concreteRequests, self).__init__()
可能毫无用处,因为 requests 中没有任何 class 声明,只是导入...
那么,requests包可以subclassed/specialized通过继承吗?
requests
是 python 模块而不是 class。你只能子class classes.
所以基本上你应该在你自己的内部利用它的 methods/functions 自定义 class.
import requests
class MyRequests:
def __init__(self, username, passwd):
self.username = username
self.passwd = passwd
def get(self, *args, **kwargs):
# do your thing
resp = requests.get(...)
# do more processing
我上面写的只是一个让你开始的例子。
我想给你一个完整的例子:
# ----example.py----
from locust import HttpUser, task, between
import requests
# Create fake api endpoint POST https://mocki.io/fake-json-api
class MyClient():
def __init__(self, host):
self.session = requests.session()
self.host = host
def create_user(self, *args, **kwargs):
return self.session.post('{}/v1/9b021e1a'.format(self.host), json={"name":"something"})
class QuickstartUser(HttpUser):
wait_time = between(1, 2)
@task
def api(self):
my_client = MyClient('https://api.mocki.io')
my_client.session = self.client
print(my_client.create_user().status_code)
print('Executing client normally')
client = MyClient('https://api.mocki.io')
print(client.create_user().status_code)
您可以执行客户端代码:
python example.py
你可以用locust执行客户端:
locust --host https://api.mocki.io --users 5 --spawn-rate 1 --locustfile example.py
在两次执行中,您使用的是同一个客户端。
一个好的方法是从请求中继承 Session
object。一个基本的例子:
from requests import Session
class MyClient(Session):
"""Specialized client that inherits the requests api."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def get_google(self):
return self.get("http://google.com")
MyClient
免费包含 Session (request
) api,以及您想要添加的任何其他内容。
A real-world 示例:假设客户端需要在运行时指定的身份验证 header(在这种情况下,身份验证需要当前时间戳)。这是一个子类 Session
和子类 AuthBase
的示例客户端,它实现了这一点(此代码需要为 api_key、secret_key、密码短语设置值):
import json, hmac, hashlib, time, requests, base64
from requests.auth import AuthBase
from requests import Session
class MyClient(Session):
"""Client with specialized auth required by api."""
def __init__(self, api_key, secret_key, passphrase, *args, **kwargs):
# allow passing args to `Session.__init__`
super().__init__(*args, **kwargs)
# `self.auth` callable that creates timestamp when request is made
self.auth = MyAuth(api_key, secret_key, passphrase)
class MyAuth(AuthBase):
"""Auth includes current timestamp when called.
https://docs.python-requests.org/en/master/user/advanced/#custom-authentication
"""
def __init__(self, api_key, secret_key, passphrase):
self.api_key = api_key
self.secret_key = secret_key
self.passphrase = passphrase
def __call__(self, request):
timestamp = str(time.time())
message = timestamp + request.method + request.path_url + (request.body or "")
message = message.encode("utf-8")
hmac_key = base64.b64decode(self.secret_key)
signature = hmac.new(hmac_key, message, hashlib.sha256)
signature_b64 = base64.b64encode(signature.digest())
request.headers.update(
{
"ACCESS-SIGN": signature_b64,
"ACCESS-TIMESTAMP": timestamp,
"ACCESS-KEY": self.api_key,
"ACCESS-PASSPHRASE": self.passphrase,
"Content-Type": "application/json",
}
)
return request