code: def _validate_workflow(
      flow, graph, flow_datastore, metadata, workflow_timeout
  ):
      seen = set()
      for var, param in flow._get_parameters():
          # Throw an exception if the parameter is specified twice.
          norm = param.name.lower()
          if norm in seen:
              raise MetaflowException(
                  "Parameter *%s* is specified twice. "
                  "Note that parameter names are "
                  "case-insensitive." % param.name
              )
          seen.add(norm)
          if "default" not in param.kwargs:
              raise MetaflowException(
                  "Parameter *%s* does not have a default value. "
                  "A default value is required for parameters when deploying flows on Airflow."
                  % param.name
              )
      # check for other compute related decorators.
      _validate_foreach_constraints(graph)
      for node in graph:
          if node.parallel_foreach:
              raise AirflowException(
                  "Deploying flows with @parallel decorator(s) "
                  "to Airflow is not supported currently."
              )
          if any([d.name == "batch" for d in node.decorators]):
              raise NotSupportedException(
                  "Step *%s* is marked for execution on AWS Batch with Airflow which isn't currently supported."
                  % node.name
              )
      SUPPORTED_DATASTORES = ("azure", "s3", "gs")
      if flow_datastore.TYPE not in SUPPORTED_DATASTORES:
          raise AirflowException(
              "Datastore type `%s` is not supported with `airflow create`. "
              "Please choose from datastore of type %s when calling `airflow create`"
              % (
                  str(flow_datastore.TYPE),
                  "or ".join(
                      ["`%s`" % x for x in SUPPORTED_DATASTORES]
                  ),
              )
          )
  
      schedule = flow._flow_decorators.get("schedule")
      if not schedule:
          return
  
      schedule = schedule[0]
      if schedule.timezone is not None:
          raise AirflowException(
              "`airflow create` does not support scheduling with `timezone`."
          )
  
imports: import base64
  import os
  import re
  import sys
  from hashlib import sha1
  from metaflow import current, decorators
  from metaflow._vendor import click
  from metaflow.exception import MetaflowException, MetaflowInternalError
  from metaflow.package import MetaflowPackage
  from metaflow.plugins.aws.step_functions.production_token import load_token, new_token, store_token
  from metaflow.plugins.kubernetes.kubernetes_decorator import KubernetesDecorator
  from metaflow.util import get_username, to_bytes, to_unicode
  from .airflow import Airflow
  from .exception import AirflowException, NotSupportedException
label_desc: Use image node_id 7 for calling a global function with example usage: _validate_workflow(flow, graph, flow_datastore, metadata, workflow_timeout) without return types