Schema

pydantic model felis.datamodel.Schema

Database schema model.

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

Parameters:

data (Any)

Fields:

• tables (Sequence[Table])

• version (SchemaVersion | str | None)

Validators:

• check_constraints » all fields

• check_tap_table_indexes » all fields

• check_unique_constraint_names » all fields

• check_unique_index_names » all fields

• check_unique_table_names » tables

• create_id_map » all fields

• generate_ids » all fields

field tables: Sequence[Table] [Required]

The tables in the schema.

Validated by:

• check_constraints

• check_description

• check_tap_table_indexes

• check_unique_constraint_names

• check_unique_index_names

• check_unique_table_names

• create_id_map

• generate_ids

field version: SchemaVersion | str | None = None

The version of the schema.

Validated by:

• check_constraints

• check_description

• check_tap_table_indexes

• check_unique_constraint_names

• check_unique_index_names

• create_id_map

• generate_ids

validator check_constraints  »  all fields

Check constraint objects for validity. This needs to be deferred until after the schema is fully loaded and the ID map is created.

Raises:

pydantic.ValidationError – Raised if any constraints are invalid.

Returns:

The schema being validated.

Return type:

Schema

Parameters:

self (Schema)

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.

validator create_id_map  »  all fields

Create a map of IDs to objects.

Returns:

The schema with the ID map created.

Return type:

Schema

Raises:

ValueError – Raised if duplicate identifiers are found in the schema.

Parameters:

self (Schema)

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

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.

• strip_ids (bool, default: False) – Whether to strip the IDs from the dumped data. Defaults to False.

Return type:

None

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

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.

• strip_ids (bool, default: False) – Whether to strip the IDs from the dumped data. Defaults to False.

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:

• resource_path (str | ParseResult | ResourcePath | Path) – The ResourcePath pointing to a YAML file.

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

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.