have a little magic #288
Closed
have a little magic #288
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
**have a little magic**
|
具体详情 请参考我的另外一个库httpx |
|
你可以在路径当中填写 @Request(method="GET",url="https://space.bilibili.com/:uid") # 或者 将:uid 替换成 {uid} 这种形式
def getUserSpacePage(response=None,**kw): # 参数最好是这种形式 或者只保留`**kw`这种形式
if(response.status_code == 200):
return response.content.decode()
homepage = getBilibiliHomePage(path={"uid":14531767}) |
|
不太理解这个修饰器作用，对于下面这个函数 bilibili-api/bilibili_api/login_func.py Lines 48 to 89 in 0181aa8 是可以改写成这样吗？ @Request(method="GET", url=API["qrcode"]["get_events"]["url"], cookies={"buvid3": str(uuid.uuid1()), "Domain": ".bilibili.com"})
def check_qrcode_events(response=None, **kw):
events = json.loads(response.content)
# 以下省略
check_qrcode_events(params={"qrcode_key": login_key}) |
|
另外这个装饰器似乎会把同步函数改成异步的？ |
因为使用的是异步的请求 |
你的理解是正确的 |
|
当你在调用的时候 就会给你的装饰函数传入一个响应对象,这是请求完了之后的response,你可以把要处理的操作放在这个函数里面 |
from dataclasses import dataclass
from inspect import iscoroutinefunction as isAsync
from typing import Any, Callable, Dict, Optional
import httpx
from aiohttp import CookieJar
@dataclass
class Request:
"请求装饰器"
url: str
method: str
timeout: Optional[int] = None
cookies: Optional[CookieJar] = None
path: Optional[Dict[str, Any]] = None
body: Optional[Dict[str, Any]] = None
form: Optional[Dict[str, Any]] = None
json: Optional[Dict[str, Any]] = None
query: Optional[Dict[str, Any]] = None
headers: Optional[Dict[str, Any]] = None
def __post_init__(self):
self.cookies = self.covernCookies(self.cookies)
def formatURLPath(self, url: str, data: Optional[Dict[str, Any]] = None):
if data is not None:
backupURL = url
for key in data:
backupURL = backupURL.replace("{%s}"%(key), str(data.get(key)), 1)
backupURL = backupURL.replace(":%s"%(key), str(data.get(key)), 1)
return backupURL
return url
def covernCookies(self, cookies: Optional[Dict[str, Any]] = None):
if cookies is not None:
# TODO: implement dict to cookiesJar
cookiejarInstance = CookieJar()
return cookiejarInstance
def update(self, **kwargs):
"更新参数"
self.url = kwargs.get("url", self.url)
self.path = kwargs.get("path", self.path)
self.method = kwargs.get("method", self.method)
overrideHeaders: Optional[Dict[str,Any]] = kwargs.get("headers")
if overrideHeaders is not None:
self.headers = self.headers or {}
for key in overrideHeaders:
self.headers[key] = overrideHeaders[key]
self.headers = kwargs.get("headers", self.headers)
self.query = kwargs.get("query", self.query)
self.json = kwargs.get("json", self.json)
def __call__(self, func: Callable):
if isAsync(func):
async def wapper(*args, **kwargs):
self.update(kwargs=kwargs)
session = get_session()
res = await session.request(
method=self.method,
url=self.formatURLPath(self.url, self.path),
json=self.json,
headers=self.headers,
params=self.query,
timeout=self.timeout
)
return await func(response=res, *args, **kwargs)
return wapper
else:
def wapper(*args, **kwargs):
self.update(kwargs=kwargs)
res = httpx.request(
method=self.method,
url=self.formatURLPath(self.url, self.path),
json=self.json,
headers=self.headers,
params=self.query,
timeout=self.timeout
)
return func(response=res, *args, **kwargs)
return wapper |
Thanks for the suggestion @Drelf2018
如何实现自动补全...不便于开发 |
|
变成 |
|
可以使用BaseModel进行自动补全 |
|
如果不提供更改注解当中设置好的东西,会导致直接就把参数写死,所以后续调用的时候定义好的东西都是可以更改的,包括方法,url等等,如果想活得类型自动补全 完全可以定义继承自 |
|
给个例子吧，我感觉用处就是：需要先请求别的接口获取数据后，再利用返回值请求另一个接口。 |
其实你也完全可以在封装好的函数里面处理完了 直接返回结果的 |
|
暂时没空处理这个... |
|
可以merge，但是可能用处不大 |
|
@z0z0r4 把这个merge了然后发release吧 |
|
Style Format @dataclass
class Request:
'''
Request 装饰器
用于代替直接调用异步函数 request
'''
url: str
method: str
path: Optional[Dict[str, Any]] = None
query: Optional[Dict[str, Any]] = None
body: Optional[Dict[str, Any]] = None
form: Optional[Dict[str, Any]] = None
cookies: Optional['CookieJar'] = None
json: Optional[Dict[str, Any]] = None
timeout: Optional[int] = None
headers: Optional[Dict[str, Any]] = None
def format_url_path(self, url: str, data: Optional[Dict[str, Any]] = None):
'''
格式化 URL
'''
if data is not None:
backup_url = url
for key in data:
backup_url = backup_url.replace(
"{%s}" % (key), str(data.get(key)), 1)
backup_url = backup_url.replace(
":%s" % (key), str(data.get(key)), 1)
return backup_url
return url
def convert_Cookies(self, cookies: Optional[Dict[str, Any]] = None):
'''
将 Cookie 转换为 CookieJar
'''
if cookies != None:
# TODO: implement dict to cookieJar
cookiejarInstance = CookieJar()
return cookiejarInstance
def update(self, **kwargs):
'''
更新参数
'''
self.url = kwargs.get("url", self.url)
self.path = kwargs.get("path", self.path)
self.method = kwargs.get("method", self.method)
override_headers: Optional[Dict[str, Any]] = kwargs.get("headers")
if override_headers is not None:
self.headers = self.headers or {}
for key in override_headers:
self.headers[key] = override_headers[key]
self.query = kwargs.get("query", self.query)
self.json = kwargs.get("json", self.json)
def __call__(self, func: Callable):
if isAsync(func):
async def wapperAsync(*args, **kwargs):
session = get_session()
self.update(kwargs=kwargs)
if self.method == "ws" or self.method == "websocket":
with connect(self.url) as websocket:
websocket.send(self.body or self.json or "")
res = websocket.recv()
res = await session.request(
method=self.method,
url=self.format_url_path(self.url, self.path),
json=self.json,
headers=self.headers,
params=self.query,
timeout=self.timeout
)
return func(response=res, *args, **kwargs)
return wapperAsync
else:
def wapper(*args, **kwargs):
self.update(kwargs=kwargs)
if self.method == "ws" or self.method == "websocket":
with connect(self.url) as websocket:
websocket.send(self.body or self.json or "")
res = websocket.recv()
res = httpx.request(
method=self.method,
url=self.format_url_path(self.url, self.path),
json=self.json,
headers=self.headers,
params=self.query,
timeout=self.timeout
)
return func(response=res, *args, **kwargs)
return **wapper**虽然我实在是不觉得把 events_api = API["qrcode"]["get_events"]
params = {"qrcode_key": login_key}
events = json.loads(
requests.get(
events_api["url"],
params=params,
cookies={"buvid3": str(uuid.uuid1()), "Domain": ".bilibili.com"},
)变成 @Request(method="GET",
url=API["qrcode"]["get_events"]["url"],
cookies={"buvid3": str(uuid.uuid1()), "Domain": ".bilibili.com"}
)对于开发有多大意义，并没有怎么方便到...我大概不会去用装饰器 @Drelf2018 倒不如试试看怎么自动把 |
感觉没法 merged，我是不是应该先发 release 火速把参数排序给处理了先 =~= |
|
今晚发 |
|
可以把这里的 |
我期末考，你看看吧... |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
来点魔法
@Request装饰器