def _get_schema()

in migration_toolkit/sql_generators/create_table/discover_result_parser.py [0:0]

• 18 lines of code
• 4 McCabe index (conditional complexity)

  def _get_schema(self, schema_name: str):
    schema = next(
        (
            d.get(self.SOURCE_TYPE_TO_TABLES[self.source_type], None)
            for d in self.discover_result
            if d[self.SOURCE_TYPE_TO_SCHEMA[self.source_type]] == schema_name
        ),
        None,
    )

    if not schema:
      available_schemas = self.list_schemas()
      raise KeyError(
          f"Source database `{schema_name}` does not appear in the discover"
          " result, or the database has no tables. Make sure the database is"
          " present in the connection profile. Available databases are:"
          f" {available_schemas}"
      )

    return schema