Schema

pydantic model felis.datamodel.Schema

Database schema model.

This represents a database schema, which contains one or more tables.

Parameters:

data (Any)

Fields:
Validators:
field tables: Sequence[Table] [Required]

The tables in the schema.

Validated by:
field version: SchemaVersion | str | None = None

The version of the schema.

Validated by:
validator check_tap_table_indexes  »  all fields

Check that the TAP table indexes are unique.

Parameters:

info (ValidationInfo) – The validation context used to determine if the check is enabled.

Returns:

The schema being validated.

Return type:

Schema

validator check_unique_constraint_names  »  all fields

Check for duplicate constraint names in the schema.

Returns:

The schema being validated.

Return type:

Schema

Raises:

ValueError – Raised if duplicate constraint names are found in the schema.

Parameters:

self (Schema)

validator check_unique_index_names  »  all fields

Check for duplicate index names in the schema.

Returns:

The schema being validated.

Return type:

Schema

Raises:

ValueError – Raised if duplicate index names are found in the schema.

Parameters:

self (Schema)

validator check_unique_table_names  »  tables

Check that table names are unique.

Parameters:

tables (list[Table]) – The tables to check.

Returns:

The tables if they are unique.

Return type:

list [ Table ]

Raises:

ValueError – Raised if table names are not unique.

dump_json(stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Pretty print the schema as JSON.

Parameters:

stream (IO[str], default: <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>) – The stream to write the JSON data to.

Return type:

None

dump_yaml(stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Pretty print the schema as YAML.

Parameters:

stream (IO[str], default: <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>) – The stream to write the YAML data to.

Return type:

None

find_object_by_id(id, obj_type)

Find an object with the given type by its ID.

Parameters:
  • id (str) – The ID of the object to find.

  • obj_type (type[TypeVar(T, bound= BaseObject)]) – The type of the object to find.

Returns:

The object with the given ID and type.

Return type:

BaseObject

Raises:
  • KeyError – If the object with the given ID is not found in the schema.

  • TypeError – If the object that is found does not have the right type.

Notes

The actual return type is the user-specified argument T, which is expected to be a subclass of BaseObject.

classmethod from_stream(source, context={})

Load a Schema from a file stream which should contain YAML data.

Parameters:
  • source (IO[str]) – The file stream to read from.

  • context (dict[str, Any], default: {}) – Pydantic context to be used in validation.

Returns:

The Felis schema loaded from the stream.

Return type:

Schema

Raises:
  • yaml.YAMLError – Raised if there is an error loading the YAML file.

  • pydantic.ValidationError – Raised if the schema fails validation.

classmethod from_uri(resource_path, context={})

Load a Schema from a string representing a ResourcePath.

Parameters:
Returns:

The ID of the object.

Return type:

str

Raises:
  • yaml.YAMLError – Raised if there is an error loading the YAML data.

  • ValueError – Raised if there is an error reading the resource.

  • pydantic.ValidationError – Raised if the schema fails validation.

validator generate_ids  »  all fields

Generate IDs for objects that do not have them.

Parameters:
  • values (dict[str, Any]) – The values of the schema.

  • info (ValidationInfo) – Validation context used to determine if ID generation is enabled.

Returns:

The values of the schema with generated IDs.

Return type:

dict [ str, Any ]

get_table_by_column(column)

Find the table that contains a column.

Parameters:

column (Column) – The column to find.

Returns:

The table that contains the column.

Return type:

Table

Raises:

ValueError – If the column is not found in any table.