0

Hope this is the right StackExchange community to ask this question.

I am building a (Python) project that will have a list if URLs to hit. They have to be hit serially, and part of response of one API will be input to the next one. I say "part of" because which part of the response is to be fed to the next input will be different for each API.

What I have thought till now is this -

  • using requests (not very keen for aio as requests will be serial)
  • using some kind of serialize-deserialze module
  • have a config that holds list of entries like
    • url
    • headers
    • response body transformation for the next API
    • request body transformation (that includes previous request's transformed response plus some additional data if required)
  • store the transformed response in a top-level variable to be returned in the end
  • main module to simply call this config in order and make the data flow till the end
    • for each request, it will check its request transformer and form the body accordingly (data for this can be taken from the variable mentioned above and/or from the transformer itself)
    • it will then hit the API and transform the response as per response transformer and store in the above variable
    • repeat for the next API
  • stop the flow if any API returns non-2xx response (business requirement)
  • stack: only requires Python, no db required, as it will be a script to be called by another one

Can any one suggest a better approach, or a library that can be helpful here?

Speed is not really the top-most priority, but intuitive design is.

Thanks!

2
  • Is the list of URls supposed to be entirely configurable, or can it be hard-coded?
    – amon
    Feb 26 at 11:09
  • endpoints are hard-coded, along with what headers or request body to pass; but these may have to be made up using responses of previous URIs', which in turn can also be hard-coded - like i need value of ["data"]["shipment"]["status"] from the previous API to be passed as a value in header or body of the next request..
    – Shod
    Feb 26 at 16:20

1 Answer 1

2

In situations like these, it's perhaps better to just write the code without any fancy architecture. Something like this:

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)

Specific notes about your points:

requests vs aio – I agree, going async has no value whatsoever if doing these sequential requests is all that the program will be doing. However, async requests would have advantages if you can perform some requests concurrently, or if this chain of requests is part of a library that should be used as part of another program.

special deserialization modules – probably overkill, especially if you're just going to use JSON. Sure, they might simplify that part of the code. But sometimes, adding another library is more complicated than just writing the necessary code directly.

driving the request chain via configuration instead of code – the problem here is that any sufficiently advanced configuration system is indistinguishable from just writing the Python code directly (compare the inner-platform effect and Greenspun's tenth rule). Such a configuration system can be worth the effort if the configuration is supposed to be provided by non-programmers, but as you say the URLs and the response transformations are unlikely to change. Configuration can also be worth it for interaction with other programs, e.g. if the progress should be displayed in a GUI. The major downside of creating such a system is that the data flow has become much more indirect and more difficult to understand.

using top-level variables – it's generally considered to be a better design to avoid mutable global variables, and to instead create helper functions that return the relevant data. This makes the data flows throughout your code more obvious, and thus makes the program easier to understand, extend, and debug.

stopping the flow on errors – that is entirely sensible, for example by throwing exceptions in the code. In the above code example, the requests.Response.raise_for_status() method is used to raise on 4xx and 5xx status codes.

no database – it is entirely feasible to keep the entire state in memory, i.e. just in Python variables. However, persisting intermediate results could be worth it if the chain takes fairly long to complete or is otherwise costly, and if you are concerned about spurious errors. For example, if the server for the last request is temporarily offline, the chain would fail and you would have to restart the program from the start. A local database such as SQLite could fix this, though a couple of JSON files would probably be sufficient. Adding such checkpoints would be easy for a configuration-driven chain, but it should be reasonably straightforward for normal code as well. For example, I might write a decorator like this (untested):

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():
  ...
1
  • thanks @amon, i'll keep these things in mind..
    – Shod
    Feb 28 at 13:13

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.