def _to_schema()

in migration_toolkit/sql_generators/copy_rows/ddl_parser.py [0:0]

• 17 lines of code
• 5 McCabe index (conditional complexity)

  def _to_schema(ddl: List[str]) -> Dict[str, BigQueryType]:
    ddl = [line.strip() for line in ddl]
    columns_start_index = ddl.index("(")

    try:
      columns_end_index = ddl.index(")")
    except ValueError:
      try:
        columns_end_index = ddl.index(");")
      except ValueError:
        raise ValueError(
            "Couldn't parse schema file. Make sure this file is generated by"
            " the response of `fetch_bigquery_table_schema.py`"
        )

    if "PRIMARY KEY" in ddl[columns_end_index - 1]:
      columns_end_index -= 1

    schema = ddl[columns_start_index + 1 : columns_end_index]

    return DDLParser._to_dict(schema)