code: def gen_function(func: swagger_parser.Function) -> Code:
      # Function name.
      out = [f"def {func.operation_name_sc()}("]
  
      # Function parameters.
      out += ['    session: "api.Session",']
      if func.params:
          out += ["    *,"]
  
      required = sorted(
          (k, v) for k, v in func.params.items() if v.required
      )
      optional = sorted(
          (k, v) for k, v in func.params.items() if not v.required
      )
  
      for _, param in required + optional:
          out += [gen_function_param(param)]
  
      # Function return type.
      # We wrap the return type annotation for streaming or union responses.
      need_quotes = func.streaming or len(func.responses) > 1
      if func.streaming:
          func.responses = {
              code: unwrap_streaming_response(r)
              for code, r in func.responses.items()
          }
      returntypes = set(
          annotation(r, prequoted=need_quotes)
          for r in func.responses.values()
      )
      returntypestr = ",".join(sorted(returntypes))
      if len(returntypes) > 1:
          returntypestr = f"typing.Union[{returntypestr}]"
      if func.streaming:
          returntypestr = f"typing.Iterable[{returntypestr}]"
      if need_quotes:
          returntypestr = f'"{returntypestr}"'
  
      out += [f") -> {returntypestr}:"]
      out += [
          TAB + line if line else ""
          for line in gen_function_docstring(func)
      ]
  
      # Function body.
      has_path_params = any(
          p for p in func.params.values() if p.where == "path"
      )
      # body_params = sorted(p for p in self.params if func.params[p].where == "body") # not in use
      query_params = sorted(
          (k, v) for k, v in func.params.items() if v.where == "query"
      )
  
      pathstr = f'"{func.path}"'
      if has_path_params:
          # Happily, we can just generate an f-string based on the path swagger gives us.
          pathstr = "f" + pathstr
  
      if query_params:
          out += ["    _params = {"]
          for _, param in query_params:
              if isinstance(param.type, swagger_parser.Bool):
                  value = f"str({param.name}).lower()"
                  if not param.required:
                      value += f" if {param.name} is not None else None"
              elif need_parse(param.type):
                  value = f"{dump(param.type, param.name, omit_unset='True')}"
                  if not param.required:
                      value += f" if {param.name} is not None else None"
              else:
                  value = param.name
              out += [f'        "{param.serialized_name}": {value},']
          out += ["    }"]
      else:
          out += ["    _params = None"]
  
      for n, param in func.params.items():
          if param.where == "path" and isinstance(
              param.type, swagger_parser.String
          ):
              # bindings string types are not enforced, so check the actual type.
              out += [f"    if type({param.name}) == str:"]
              out += [
                  f"        {param.name} = parse.quote({param.name})"
              ]
  
      if "body" in func.params:
          # It is important that request bodies omit unset values so that PATCH request bodies
          # do not include extraneous None values.
          body_param = func.params["body"]
          bodystr = dump(body_param.type, body_param.name, "True")
      else:
          bodystr = "None"
      out += ["    _resp = session._do_request("]
      out += [f'        method="{func.method.upper()}",']
      out += [f"        path={pathstr},"]
      out += ["        params=_params,"]
      out += [f"        json={bodystr},"]
      out += ["        data=None,"]
      out += ["        headers=None,"]
      out += ["        timeout=None,"]
      out += [f"        stream={func.streaming},"]
      out += ["    )"]
      for expect, returntype in func.responses.items():
          out += [f"    if _resp.status_code == {expect}:"]
          is_none = isinstance(returntype, swagger_parser.Ref) and (
              returntype.defn is None
              or isinstance(returntype.defn, swagger_parser.Class)
              and not returntype.defn.params
          )
          if not func.streaming:
              if is_none:
                  out += ["        return"]
              else:
                  out += [
                      f'        return {load(returntype, "_resp.json()")}'
                  ]
          else:
              assert (
                  not is_none
              ), "unable to stream empty result class: {func}"
              # Too many quotes to do it inline:
              yieldable = load(returntype, '_j["result"]')
              out += [
                  f"        try:",
                  f"            for _line in _resp.iter_lines(chunk_size=1024 * 1024):",
                  f"                _j = json.loads(_line)",
                  f'                if "error" in _j:',
                  f"                    raise APIHttpStreamError(",
                  f'                        "{func.operation_name_sc()}",',
                  f'                        runtimeStreamError.from_json(_j["error"])',
                  f"                )",
                  f"                yield {yieldable}",
                  f"        except requests.exceptions.ChunkedEncodingError:",
                  f'            raise APIHttpStreamError("{func.operation_name_sc()}", runtimeStreamError(message="ChunkedEncodingError"))',
                  f"        return",
              ]
      out += [
          f'    raise APIHttpError("{func.operation_name_sc()}", _resp)'
      ]
  
      return "\n".join(out)
  
imports: import os
  import typing
  import swagger_parser
  from typing_extensions import assert_never
label_desc: Use image node_id 12 for calling a global function with example usage: gen_function(func) and returns: str