def load()

in qpid/spec08.py [0:0]

• 66 lines of code
• 19 McCabe index (conditional complexity)

def load(specfile, *errata):
  doc = mllib.xml_parse(specfile)
  spec_root = doc["amqp"]
  spec = Spec(int(spec_root["@major"]), int(spec_root["@minor"]), specfile)

  for root in [spec_root] + [mllib.xml_parse(x)["amqp"] for x in errata]:
    # constants
    for nd in root.query["constant"]:
      val = nd["@value"]
      if val.startswith("0x"): val = int(val, 16)
      else: val = int(val)
      const = Constant(spec, pythonize(nd["@name"]), val, nd["@class"],
                       get_docs(nd))
      try:
        spec.constants.add(const)
      except ValueError as e:
        pass
        #print "Warning:", e

    # domains are typedefs
    structs = []
    for nd in root.query["domain"]:
      type = nd["@type"]
      if type == None:
        st_nd = nd["struct"]
        code = st_nd["@type"]
        if code not in (None, "", "none"):
          code = int(code)
        type = Struct(width(st_nd["@size"]), code, width(st_nd["@pack"], 2))
        if type.type != None:
          spec.structs[type.type] = type
        structs.append((type, st_nd))
      else:
        type = pythonize(type)
      domain = Domain(spec, pythonize(nd["@name"]), type, get_desc(nd),
                      get_docs(nd))
      spec.domains.add(domain)

    # structs
    for st, st_nd in structs:
      load_fields(st_nd, st.fields, spec.domains.byname)

    # classes
    for c_nd in root.query["class"]:
      cname = pythonize(c_nd["@name"])
      if cname in spec.classes.byname:
        klass = spec.classes.byname[cname]
      else:
        klass = Class(spec, cname, int(c_nd["@index"]), c_nd["@handler"],
                      get_docs(c_nd))
        spec.classes.add(klass)

      added_methods = []
      load_fields(c_nd, klass.fields, spec.domains.byname)
      for m_nd in c_nd.query["method"]:
        mname = pythonize(m_nd["@name"])
        if mname in klass.methods.byname:
          meth = klass.methods.byname[mname]
        else:
          meth = Method(klass, mname,
                        int(m_nd["@index"]),
                        m_nd["@content"] == "1",
                        [pythonize(nd["@name"]) for nd in m_nd.query["response"]],
                        get_result(m_nd, spec),
                        m_nd["@synchronous"] == "1",
                        get_desc(m_nd),
                        get_docs(m_nd))
          klass.methods.add(meth)
          added_methods.append(meth)
        load_fields(m_nd, meth.fields, spec.domains.byname)
      # resolve the responses
      for m in added_methods:
        m.responses = [klass.methods.byname[r] for r in m.responses]
        for resp in m.responses:
          resp.response = True

  spec.post_load()
  return spec