首页
学习
活动
专区
圈层
工具
发布
社区首页 >问答首页 >设计一个能够串行访问多个HTTP请求的系统

设计一个能够串行访问多个HTTP请求的系统
EN

Software Engineering用户
提问于 2022-02-26 08:21:00
回答 1查看 1.2K关注 0票数 0

希望这是正确的StackExchange社区来问这个问题。

我正在构建一个(Python)项目,如果要点击URL,它将有一个列表。它们必须按顺序命中,并且一个API的部分响应将被输入到下一个API。我说“部分”,因为响应的哪一部分将被输入到下一个输入中,对于每个API来说都是不同的。

到现在为止我想的是-

  • 使用requests (不太热衷于aio,因为请求将是串行的)
  • 使用某种序列化反序列化模块
    • jsonbender是我一直在使用的选项
    • 棉花糖也是一种选择,但它是否过分了呢?

  • 拥有一个包含像config这样的条目列表的
    • url
    • headers
    • 下一个API的响应体转换
    • 请求体转换(这包括以前请求的转换响应,如果需要的话,再加上一些额外的数据)

  • 将转换后的响应存储在将在最后返回的顶级变量中。
  • 主模块简单地按顺序调用这个config,并使数据流一直到结束。
    • 对于每个请求,它将检查其请求转换器并相应地组成主体(相关数据可以从上述变量和/或变压器本身获取)。
    • 然后,它将命中API并按照响应转换器转换响应并存储在上述变量中。
    • 对下一个API重复使用

  • 如果任何API返回非2xx响应(业务需求),则停止流。
  • 堆栈:只需要Python,不需要db,因为它将是将由另一个脚本调用的脚本

有谁能提出一个更好的方法,或者一个在这里有帮助的库吗?

速度并不是最重要的,但直观的设计是。

谢谢!

EN

回答 1

Software Engineering用户

发布于 2022-02-27 10:57:16

在这种情况下,最好只是编写代码而不使用任何花哨的体系结构。就像这样:

代码语言:javascript
复制
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文件可能就足够了。添加这样的检查点对于配置驱动的链来说是很容易的,但是对于普通代码来说,它也应该是相当简单的。例如,我可能会编写这样的装饰器(未经测试):

代码语言:javascript
复制
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():
  ...
票数 2
EN
页面原文内容由Software Engineering提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接:

https://softwareengineering.stackexchange.com/questions/437015

复制
相关文章

相似问题

领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档