希望这是正确的StackExchange社区来问这个问题。
我正在构建一个(Python)项目,如果要点击URL,它将有一个列表。它们必须按顺序命中,并且一个API的部分响应将被输入到下一个API。我说“部分”,因为响应的哪一部分将被输入到下一个输入中,对于每个API来说都是不同的。
到现在为止我想的是-
requests (不太热衷于aio,因为请求将是串行的)config这样的条目列表的urlheadersconfig,并使数据流一直到结束。有谁能提出一个更好的方法,或者一个在这里有帮助的库吗?
速度并不是最重要的,但直观的设计是。
谢谢!
发布于 2022-02-27 10:57:16
在这种情况下,最好只是编写代码而不使用任何花哨的体系结构。就像这样:
def main():
a = get_a()
b = get_b(a)
def get_a():
res = requests.get("https://example.com/a")
res.raise_for_status()
data = res.json()
return data["a"]
def get_b(a):
res = requests.get("https://example.com/b", params={"a": a})
res.raise_for_status()
data = res.json()
# "y" field is optional
return data["x"] + data.get("y", 0)关于你的观点的具体说明:
请求与aio -我同意,如果执行这些顺序的请求是程序将要做的所有事情,则去异步没有任何价值。但是,如果您可以同时执行一些请求,或者如果这个请求链是应该作为另一个程序的一部分的库的一部分,则异步请求将具有优势。
特殊的反序列化模块--可能有些过分,特别是当您只使用JSON的时候。当然,它们可以简化代码的这一部分。但有时,添加另一个库比直接编写必要的代码要复杂得多。
通过配置(而不是通过代码)驱动请求链--这里的问题是,任何足够高级的配置系统都无法区别于直接编写Python代码(比较内平台效应和格林斯夫第十法则)。如果配置应该由非程序员提供,那么这样的配置系统是值得的,但是正如您所说的,URL和响应转换不太可能改变。对于与其他程序的交互,配置也是值得的,例如,如果进度应该显示在GUI中。创建这样一个系统的主要缺点是数据流变得更加间接和难以理解。
使用顶级变量通常被认为是一个更好的设计,以避免可变的全局变量,而不是创建返回相关数据的助手函数。这使得整个代码中的数据流更加明显,从而使程序更易于理解、扩展和调试。
停止错误流--这是完全合理的,例如,在代码中抛出异常。在上面的代码示例中,使用requests.Response.raise_for_status()方法在4xx和5xx状态代码上引发。
没有数据库--将整个状态保存在内存中是完全可行的,即只在Python变量中。但是,如果链需要相当长的时间来完成,或者以其他方式代价高昂,如果您担心虚假的错误,那么持久化的中间结果可能是值得的。例如,如果上一次请求的服务器暂时脱机,则链将失败,您必须从一开始就重新启动程序。像SQLite这样的本地数据库可以修复这个问题,尽管几个JSON文件可能就足够了。添加这样的检查点对于配置驱动的链来说是很容易的,但是对于普通代码来说,它也应该是相当简单的。例如,我可能会编写这样的装饰器(未经测试):
import functools, json, hashlib, glob, os
def clear_cache(stem = "checkpoint"):
for cache_file in glob.glob(f"{stem}.*.json"):
os.remove(cache_file)
def sha256_json_fingerprint(data):
data_as_json = json.dumps(data, sort_keys=True)
return hashlib.sha256(data_as_json.encode()).hexdigest()
def cached(fn, *, name = ""):
name = name or fn.__name__
@functools.wraps(fn)
def wrapped_fn(*args, **kwargs):
# assuming all args and kwargs can be converted to JSON
key = sha256_json_fingerprint([args, kwargs])
filename = f"checkpoint.{name}.{key}.json"
# try loading the cached result
try:
with open(filename) as f:
return json.load(f)
except OsError:
pass
# remove outdated files
clear_cache(f"checkpoint.{name}")
# fall back to executing the target function
result = fn(*args, **kwargs)
with open(filename, "w") as f:
json.dump(result, f)
return result
return wrapped_fn
@cached
def get_a():
...https://softwareengineering.stackexchange.com/questions/437015
复制相似问题