Skip to content

Index

Catalog dataclass

A class representing a Unity Catalog - Catalog.

Source code in src/cloe_nessy/models/catalog.py 
 4
 5
 6
 7
 8
 9
10
@dataclass
class Catalog:
    """A class representing a Unity Catalog - Catalog."""

    name: str
    owner: str = ""
    comment: str = ""

Column

Bases: BaseModel

Represents a Column of a Table.

Source code in src/cloe_nessy/models/column.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
class Column(BaseModel):
    """Represents a Column of a Table."""

    name: str
    data_type: str
    nullable: bool = True
    default_value: Any = None
    generated: str | None = None
    properties: dict[str, Any] = Field(default_factory=dict)
    comment: str | None = None

    @field_validator("data_type", mode="before")
    def data_type_transform(cls, raw: str) -> str:
        """Map potential aliases to the correct SQL data type.

        Args:
            raw: The value for the data type.
        """
        val = raw.lower()
        base_data_types = re.findall(r"\b[a-z]+\b", val)
        forbidden_characters = re.findall(r"[^a-z\<\>)]+", val)

        if forbidden_characters:
            raise ValueError(f"Forbidden characters in data type definition [ '{val}' ]: [' {forbidden_characters} ']")
        for base_data_type in base_data_types:
            if base_data_type not in COLUMN_DATA_TYPE_LIST:
                raise ValueError(f"Unknown data type used in data type definition [ '{val}' ]")
        return val

    @model_validator(mode="before")
    def _validate_generated_and_default_value(cls, v: Any) -> Any:
        """Check if a column has a default value and is generated.

        That doesn't make sense, so an error should be raised.
        """
        if v.get("default_value") and v.get("generated"):
            raise ValueError("A column can't have a default value and be generated.")
        if (v.get("default_value") or v.get("generated")) and v.get("nullable") is True:
            raise ValueError("A column can't have a default value or be generated and be nullable.")
        return v

_validate_generated_and_default_value(v)

Check if a column has a default value and is generated.

That doesn't make sense, so an error should be raised.

Source code in src/cloe_nessy/models/column.py 
55
56
57
58
59
60
61
62
63
64
65
@model_validator(mode="before")
def _validate_generated_and_default_value(cls, v: Any) -> Any:
    """Check if a column has a default value and is generated.

    That doesn't make sense, so an error should be raised.
    """
    if v.get("default_value") and v.get("generated"):
        raise ValueError("A column can't have a default value and be generated.")
    if (v.get("default_value") or v.get("generated")) and v.get("nullable") is True:
        raise ValueError("A column can't have a default value or be generated and be nullable.")
    return v

data_type_transform(raw)

Map potential aliases to the correct SQL data type.

Parameters:

Name Type Description Default
raw str

The value for the data type.

 required
Source code in src/cloe_nessy/models/column.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@field_validator("data_type", mode="before")
def data_type_transform(cls, raw: str) -> str:
    """Map potential aliases to the correct SQL data type.

    Args:
        raw: The value for the data type.
    """
    val = raw.lower()
    base_data_types = re.findall(r"\b[a-z]+\b", val)
    forbidden_characters = re.findall(r"[^a-z\<\>)]+", val)

    if forbidden_characters:
        raise ValueError(f"Forbidden characters in data type definition [ '{val}' ]: [' {forbidden_characters} ']")
    for base_data_type in base_data_types:
        if base_data_type not in COLUMN_DATA_TYPE_LIST:
            raise ValueError(f"Unknown data type used in data type definition [ '{val}' ]")
    return val

Constraint

Bases: BaseModel

Represents a Constraint on a Table.

Source code in src/cloe_nessy/models/constraint.py 
4
5
6
7
8
9
class Constraint(BaseModel):
    """Represents a Constraint on a Table."""

    name: str
    expression: str
    description: str | None = None

ForeignKey

Bases: BaseModel

Represents a ForeignKey.

Source code in src/cloe_nessy/models/foreign_key.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class ForeignKey(BaseModel):
    """Represents a ForeignKey."""

    foreign_key_columns: list[str]
    parent_table: str
    parent_columns: list[str]
    foreign_key_option: list[str] | None = None

    @field_validator("foreign_key_columns", mode="before")
    def _validate_foreign_key_columns(cls, v):
        return _process_column_input(v)

    @field_validator("parent_columns", mode="before")
    def _validate_parent_columns(cls, v):
        return _process_column_input(v)

    @field_validator("parent_table", mode="before")
    def _validate_identifier(cls, v):
        if len(v.split(".")) != 3:
            raise ValueError("The 'parent_table' must be in the format 'catalog.schema.table'")
        if "<env>" in v:
            v = v.replace("<env>", os.environ["PROJECT_ENVIRONMENT"])
        return v

Schema

Bases: ReadInstancesMixin

A Class to represent a Schema in Unity Catalog.

Source code in src/cloe_nessy/models/schema.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class Schema(ReadInstancesMixin):
    """A Class to represent a Schema in Unity Catalog."""

    catalog: str
    name: str
    storage_path: str | None = None
    tables: list[Table] = Field(default_factory=list)
    properties: dict[str, Any] = Field(default_factory=dict)
    comment: str | None = None

    @classmethod
    def read_instance_from_file(
        cls,
        instance_path: str | Path,
        fail_on_missing_subfolder: bool = True,
        table_dir_name: str = "tables",
        **_: Any,
    ) -> tuple[Self | None, list[ValidationErrorType]]:
        """Read a schema from file.

        Adds the table objects from a path relative to the schema definition.

        Args:
            instance_path: The path to the Schema definition YAML file.
            fail_on_missing_subfolder: If False return a tuple with 2 empty
                    lists. Otherwise raise a FileNotFoundError.
            table_dir_name: The name of the directory containing the Table
                    definitions related to this schema. Can be a relative path.
        """
        processed_instance_path = process_path(instance_path)
        if not processed_instance_path:
            raise FileNotFoundError("Schema file not found.")

        schema, schema_errors = super().read_instance_from_file(processed_instance_path)
        if schema:
            schema.storage_path = "" if not schema.storage_path else schema.storage_path
            tables, table_errors = Table.read_instances_from_directory(
                instance_path=processed_instance_path.parents[0] / table_dir_name,
                catalog_name=schema.catalog,
                schema_name=schema.name,
                schema_storage_path=schema.storage_path,
                fail_on_missing_subfolder=fail_on_missing_subfolder,
            )
            schema.tables = tables
        return schema, schema_errors + table_errors

    def get_table_by_name(self, table_name: str) -> Table:
        """Return table in schema.

        Filters tables in schema for table_name and returns Table object.

        Args:
            table_name: Name of table to return from schema.

        Raises:
            ValueError: If table not found in schema metadata.

        Returns:
            The table.
        """
        table = next((table for table in self.tables if table.name == table_name), None)

        if not table:
            raise ValueError(f"Table {table_name} not found in {self.catalog}.{self.name} metadata.")

        return table

    def add_table(self, table: Table):
        """Adds a table to the schema and sets the table identifier accordingly.

        Args:
            table: A Table object that is added to the Schema tables.
        """
        table.identifier = f"{self.catalog}.{self.name}.{table.name}"
        self.tables.append(table)

    def add_tables(self, tables: list[Table]) -> None:
        """Adds tables to the schema.

        Args:
            tables: A list of Table objects that are added to the Schema tables.
        """
        for table in tables:
            self.add_table(table)

add_table(table)

Adds a table to the schema and sets the table identifier accordingly.

Parameters:

Name Type Description Default
table Table

A Table object that is added to the Schema tables.

 required
Source code in src/cloe_nessy/models/schema.py 
79
80
81
82
83
84
85
86
def add_table(self, table: Table):
    """Adds a table to the schema and sets the table identifier accordingly.

    Args:
        table: A Table object that is added to the Schema tables.
    """
    table.identifier = f"{self.catalog}.{self.name}.{table.name}"
    self.tables.append(table)

add_tables(tables)

Adds tables to the schema.

Parameters:

Name Type Description Default
tables list[Table]

A list of Table objects that are added to the Schema tables.

 required
Source code in src/cloe_nessy/models/schema.py 
88
89
90
91
92
93
94
95
def add_tables(self, tables: list[Table]) -> None:
    """Adds tables to the schema.

    Args:
        tables: A list of Table objects that are added to the Schema tables.
    """
    for table in tables:
        self.add_table(table)

get_table_by_name(table_name)

Return table in schema.

Filters tables in schema for table_name and returns Table object.

Parameters:

Name Type Description Default
table_name str

Name of table to return from schema.

 required

Raises:

Type Description
ValueError

If table not found in schema metadata.

Returns:

Type Description
Table

The table.
Source code in src/cloe_nessy/models/schema.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def get_table_by_name(self, table_name: str) -> Table:
    """Return table in schema.

    Filters tables in schema for table_name and returns Table object.

    Args:
        table_name: Name of table to return from schema.

    Raises:
        ValueError: If table not found in schema metadata.

    Returns:
        The table.
    """
    table = next((table for table in self.tables if table.name == table_name), None)

    if not table:
        raise ValueError(f"Table {table_name} not found in {self.catalog}.{self.name} metadata.")

    return table

read_instance_from_file(instance_path, fail_on_missing_subfolder=True, table_dir_name='tables', **_) classmethod

Read a schema from file.

Adds the table objects from a path relative to the schema definition.

Parameters:

Name Type Description Default
instance_path str | Path

The path to the Schema definition YAML file.

 required
fail_on_missing_subfolder bool

If False return a tuple with 2 empty lists. Otherwise raise a FileNotFoundError.

 True
table_dir_name str

The name of the directory containing the Table definitions related to this schema. Can be a relative path.

 'tables'
Source code in src/cloe_nessy/models/schema.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@classmethod
def read_instance_from_file(
    cls,
    instance_path: str | Path,
    fail_on_missing_subfolder: bool = True,
    table_dir_name: str = "tables",
    **_: Any,
) -> tuple[Self | None, list[ValidationErrorType]]:
    """Read a schema from file.

    Adds the table objects from a path relative to the schema definition.

    Args:
        instance_path: The path to the Schema definition YAML file.
        fail_on_missing_subfolder: If False return a tuple with 2 empty
                lists. Otherwise raise a FileNotFoundError.
        table_dir_name: The name of the directory containing the Table
                definitions related to this schema. Can be a relative path.
    """
    processed_instance_path = process_path(instance_path)
    if not processed_instance_path:
        raise FileNotFoundError("Schema file not found.")

    schema, schema_errors = super().read_instance_from_file(processed_instance_path)
    if schema:
        schema.storage_path = "" if not schema.storage_path else schema.storage_path
        tables, table_errors = Table.read_instances_from_directory(
            instance_path=processed_instance_path.parents[0] / table_dir_name,
            catalog_name=schema.catalog,
            schema_name=schema.name,
            schema_storage_path=schema.storage_path,
            fail_on_missing_subfolder=fail_on_missing_subfolder,
        )
        schema.tables = tables
    return schema, schema_errors + table_errors

Table

Bases: TemplateLoaderMixin, ReadInstancesMixin, LoggerMixin

A Class to represent a Table in Unity Catalog.

Source code in src/cloe_nessy/models/table.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
class Table(TemplateLoaderMixin, ReadInstancesMixin, LoggerMixin):
    """A Class to represent a Table in Unity Catalog."""

    identifier: str
    columns: list[Column]
    is_external: bool | None = None
    partition_by: list[str] = Field(default_factory=list)
    liquid_clustering: bool | None = None
    composite_primary_key: list[str] = Field(default_factory=list)
    properties: dict[str, str] = Field(default_factory=dict)
    constraints: list[Constraint] = Field(default_factory=list)
    foreign_keys: list[ForeignKey] = Field(default_factory=list)
    storage_path: str | None = None
    business_properties: dict[str, str] = Field(default_factory=dict)
    comment: str | None = None
    data_source_format: str | None = None

    def model_post_init(self, __context: Any) -> None:
        """Post init method for the Table model."""
        self._console_logger = self.get_console_logger()
        self._tabular_logger = self.get_tabular_logger(uc_table_name="nessy_table_logs", log_type="nessy_table_logs")
        self._console_logger.debug(f"Model for table [ '{self.identifier}' ] has been initialized.")
        self._tabular_logger.debug(f"Message : Model for table [ '{self.identifier}' ] has been initialized.")

    @property
    def catalog(self):
        """The name of the Catalog of the Table."""
        return self.identifier.split(".")[0]

    @property
    def schema(self):
        """The name of the Schema of the Table."""
        return self.identifier.split(".")[1]

    @property
    def name(self):
        """The name of the Table."""
        return self.identifier.split(".")[2]

    @property
    def escaped_identifier(self):
        """The escaped identifier of the Table."""
        return f"`{self.catalog}`.`{self.schema}`.`{self.name}`"

    @field_validator("constraints", mode="before")
    def _validate_constraints(cls, raw: dict[str, dict[str, str]]) -> list[Constraint]:
        """The constraints are defined with the name as key or as a list and must therefore be transformed."""
        if isinstance(raw, dict):
            constraints = [Constraint(name=constraint, **raw[constraint]) for constraint in raw]
        elif isinstance(raw, list):
            constraints = []
            for item in raw:
                if isinstance(item, Constraint):
                    constraints.append(item)
                elif isinstance(item, dict):
                    constraints.append(Constraint(**item))
                else:
                    raise ValueError("Invalid constraint format")
        else:
            raise ValueError("Constraints must be either a list or a dictionary")
        return constraints

    @field_validator("foreign_keys", mode="after")
    def _validate_fk_columns(cls, v: list[ForeignKey], values: ValidationInfo):
        """Foreign keys need to be columns in the table as well."""
        column_names = [c.name for c in values.data.get("columns", [])]
        for fk in v:
            for column in fk.foreign_key_columns:
                if column not in column_names:
                    raise ValueError(f"Foreign key column '{column}' does not match any column in 'columns'")
        return v

    @model_validator(mode="after")
    def _validate_is_external(cls, table: Self):
        """If is_external is set to True, storage_path has to be set."""
        if table.is_external and table.storage_path is None:
            raise ValueError("is_external cannot be true while storage_path is None.")
        return table

    @classmethod
    def read_instances_from_directory(
        cls,
        instance_path: str | Path,
        fail_on_missing_subfolder: bool = True,
        catalog_name: str | None = None,
        schema_name: str | None = None,
        schema_storage_path: str | Path | None = None,
        **_: Any,
    ) -> tuple[list[Self], list[ValidationErrorType]]:
        """Reads instances from a directory containing YAML files.

        This method scans a specified directory for YAML files (.yaml or .yml),
        attempts to read and parse each file as an instance of the class, and
        collects any errors encountered during the process. If the directory
        does not exist or is not a directory, it either raises a
        FileNotFoundError (if fail_on_missing_subfolder is True) or returns
        empty lists.

        Args:
            instance_path: The path to the directory containing instance files.
            catalog_name: Name of the catalog to which these instances belong.
            schema_name: Name of the schema used for validating the instances.
            fail_on_missing_subfolder: Determines behavior when the specified
                    directory does not exist or is not a directory. Defaults to True,
                    which will raise a FileNotFoundError.
            schema_storage_path: Path to the storage location of the schema
                    these tables instances belong to.

        Returns:
            - The first list contains instances of the class that were
                successfully read and validated from the files.
            - The second list contains errors encountered during the
                process, which could be validation errors or YAML
                parsing/scanning errors.

        Raises:
            FileNotFoundError: If the specified directory does not exist or is
                               not a directory and fail_on_missing_subfolder is True.
            ValueError: If catalog_name or schema_name are not provided.
        """
        processed_instance_path = process_path(instance_path)
        schema_storage_path = process_path(schema_storage_path)
        errors: list[ValidationErrorType] = []

        if not catalog_name or not schema_name:
            errors.append(ValueError("catalog_name and schema_name must be provided."))
            return [], errors
        instances: list[Self] = []

        if not processed_instance_path or not processed_instance_path.exists() or not processed_instance_path.is_dir():
            if fail_on_missing_subfolder:
                raise FileNotFoundError(f"Directory not found: {processed_instance_path}")
            else:
                return instances, errors

        for instance_file in processed_instance_path.iterdir():
            sub_errors: list[ValidationErrorType] = []
            if instance_file.is_file() and instance_file.suffix in (".yaml", ".yml"):
                instance, sub_errors = cls.read_instance_from_file(
                    instance_file, catalog_name, schema_name, str(schema_storage_path)
                )
                instances += [] if instance is None else [instance]
            errors += sub_errors

        return instances, errors

    @classmethod
    def read_instance_from_file(
        cls,
        instance_path: str | Path,
        catalog_name: str | None = None,
        schema_name: str | None = None,
        schema_storage_path: str | Path | None = None,
        **_: Any,
    ) -> tuple[Self | None, list[ValidationErrorType]]:
        """Read a table instance from file.

        Args:
            instance_path: The path to the Schema definition YAML file.
            catalog_name: The name of the Catalog of the Table.
            schema_name: The name of the Schema of the Table.
            schema_storage_path: The storage path location of the Schema of the Table.

        Returns:
            - Instance of the class that was successfully instantiated and
                  validated
            - The second list contains errors encountered during the process,
                  which could be validation errors or YAML parsing/scanning errors.

        Raises:
            FileNotFoundError: If the specified directory does not exist or is
                not a directory and fail_on_missing_subfolder is
                True.
            ValueError: If catalog_name or schema_name are not provided.
        """
        processed_instance_path = process_path(instance_path)
        if not processed_instance_path:
            raise FileNotFoundError("Table file not found.")
        schema_storage_path = process_path(schema_storage_path)
        errors: list[ValidationErrorType] = []

        if not catalog_name or not schema_name:
            errors.append(ValueError("catalog_name and schema_name must be provided."))
            return None, errors

        try:
            with processed_instance_path.open("r") as file:
                data = yaml.safe_load(file)
                data["identifier"] = f"{catalog_name}.{schema_name}.{data['name']}"
                if data.get("is_external"):
                    if storage_path := data.get("storage_path"):
                        data["storage_path"] = storage_path
                    elif schema_storage_path:
                        data["storage_path"] = (schema_storage_path / data["name"]).as_posix()
                    else:
                        raise ValueError(
                            f"Neither storage path nor schema storage path of table {data['name']} has been provided."
                        )

                instance, sub_errors = cls.metadata_to_instance(data)
                errors += sub_errors
        except (
            ValidationError,
            ParserError,
            ScannerError,
        ) as e:
            instance = None
            errors.append(e)
        return instance, errors

    def get_create_statement(
        self,
        replace: bool = True,
    ):
        """Get the create statement for the Table.

        Args:
            replace: Whether to use the REPLACE statement or not.

        Returns:
            The rendered create statement for the Table.

        Raises:
            TemplateNotFound: If the template file is not found.
        """
        templates = Path(__file__).parent / "templates"
        template_name = "create_table.sql.j2"

        try:
            template = self.get_template(templates, template_name)
        except TemplateNotFound as err:
            self._console_logger.error(f"Template [ {template_name} ] not found.")
            raise err
        render = template.render(table=self, replace=replace)
        return render

    def get_column_by_name(self, column_name: str) -> Column | None:
        """Get a column by name.

        Args:
            column_name: The name of the column to get.

        Returns:
            The column if found, else None.
        """
        for column in self.columns:
            if column.name == column_name:
                return column
        return None

    def update_column(self, column: Column) -> None:
        """Replaces a Column with a new Column object to update it.

        Args:
            column: The new column object, to replace the old one.
        """
        self.remove_column(column)
        self.add_column(column)

    def add_column(self, column: Column):
        """Adds a column to the table.

        Args:
            column: The column to be added.
        """
        self.columns.append(column)

    def remove_column(self, column: str | Column) -> None:
        """Remove a column from the Table.

        Args.
            column: The column to be removed.
        """
        if isinstance(column, Column):
            column_name = column.name
        else:
            column_name = column

        self.columns = [col for col in self.columns if col.name != column_name]

catalog property

The name of the Catalog of the Table.

escaped_identifier property

The escaped identifier of the Table.

name property

The name of the Table.

schema property

The name of the Schema of the Table.

_validate_constraints(raw)

The constraints are defined with the name as key or as a list and must therefore be transformed.

Source code in src/cloe_nessy/models/table.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@field_validator("constraints", mode="before")
def _validate_constraints(cls, raw: dict[str, dict[str, str]]) -> list[Constraint]:
    """The constraints are defined with the name as key or as a list and must therefore be transformed."""
    if isinstance(raw, dict):
        constraints = [Constraint(name=constraint, **raw[constraint]) for constraint in raw]
    elif isinstance(raw, list):
        constraints = []
        for item in raw:
            if isinstance(item, Constraint):
                constraints.append(item)
            elif isinstance(item, dict):
                constraints.append(Constraint(**item))
            else:
                raise ValueError("Invalid constraint format")
    else:
        raise ValueError("Constraints must be either a list or a dictionary")
    return constraints

_validate_fk_columns(v, values)

Foreign keys need to be columns in the table as well.

Source code in src/cloe_nessy/models/table.py 
88
89
90
91
92
93
94
95
96
@field_validator("foreign_keys", mode="after")
def _validate_fk_columns(cls, v: list[ForeignKey], values: ValidationInfo):
    """Foreign keys need to be columns in the table as well."""
    column_names = [c.name for c in values.data.get("columns", [])]
    for fk in v:
        for column in fk.foreign_key_columns:
            if column not in column_names:
                raise ValueError(f"Foreign key column '{column}' does not match any column in 'columns'")
    return v

_validate_is_external(table)

If is_external is set to True, storage_path has to be set.

Source code in src/cloe_nessy/models/table.py 
 98
 99
100
101
102
103
@model_validator(mode="after")
def _validate_is_external(cls, table: Self):
    """If is_external is set to True, storage_path has to be set."""
    if table.is_external and table.storage_path is None:
        raise ValueError("is_external cannot be true while storage_path is None.")
    return table

add_column(column)

Adds a column to the table.

Parameters:

Name Type Description Default
column Column

The column to be added.

 required
Source code in src/cloe_nessy/models/table.py 
285
286
287
288
289
290
291
def add_column(self, column: Column):
    """Adds a column to the table.

    Args:
        column: The column to be added.
    """
    self.columns.append(column)

get_column_by_name(column_name)

Get a column by name.

Parameters:

Name Type Description Default
column_name str

The name of the column to get.

 required

Returns:

Type Description
Column | None

The column if found, else None.
Source code in src/cloe_nessy/models/table.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
def get_column_by_name(self, column_name: str) -> Column | None:
    """Get a column by name.

    Args:
        column_name: The name of the column to get.

    Returns:
        The column if found, else None.
    """
    for column in self.columns:
        if column.name == column_name:
            return column
    return None

get_create_statement(replace=True)

Get the create statement for the Table.

Parameters:

Name Type Description Default
replace bool

Whether to use the REPLACE statement or not.

 True

Returns:

Type Description

The rendered create statement for the Table.

Raises:

Type Description
TemplateNotFound

If the template file is not found.
Source code in src/cloe_nessy/models/table.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def get_create_statement(
    self,
    replace: bool = True,
):
    """Get the create statement for the Table.

    Args:
        replace: Whether to use the REPLACE statement or not.

    Returns:
        The rendered create statement for the Table.

    Raises:
        TemplateNotFound: If the template file is not found.
    """
    templates = Path(__file__).parent / "templates"
    template_name = "create_table.sql.j2"

    try:
        template = self.get_template(templates, template_name)
    except TemplateNotFound as err:
        self._console_logger.error(f"Template [ {template_name} ] not found.")
        raise err
    render = template.render(table=self, replace=replace)
    return render

model_post_init(__context)

Post init method for the Table model.

Source code in src/cloe_nessy/models/table.py 
43
44
45
46
47
48
def model_post_init(self, __context: Any) -> None:
    """Post init method for the Table model."""
    self._console_logger = self.get_console_logger()
    self._tabular_logger = self.get_tabular_logger(uc_table_name="nessy_table_logs", log_type="nessy_table_logs")
    self._console_logger.debug(f"Model for table [ '{self.identifier}' ] has been initialized.")
    self._tabular_logger.debug(f"Message : Model for table [ '{self.identifier}' ] has been initialized.")

read_instance_from_file(instance_path, catalog_name=None, schema_name=None, schema_storage_path=None, **_) classmethod

Read a table instance from file.

Parameters:

Name Type Description Default
instance_path str | Path

The path to the Schema definition YAML file.

 required
catalog_name str | None

The name of the Catalog of the Table.

 None
schema_name str | None

The name of the Schema of the Table.

 None
schema_storage_path str | Path | None

The storage path location of the Schema of the Table.

 None

Returns:

Type Description
Self | None
  • Instance of the class that was successfully instantiated and validated
list[ValidationErrorType]
  • The second list contains errors encountered during the process, which could be validation errors or YAML parsing/scanning errors.

Raises:

Type Description
FileNotFoundError

If the specified directory does not exist or is not a directory and fail_on_missing_subfolder is True.
ValueError

If catalog_name or schema_name are not provided.
Source code in src/cloe_nessy/models/table.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
@classmethod
def read_instance_from_file(
    cls,
    instance_path: str | Path,
    catalog_name: str | None = None,
    schema_name: str | None = None,
    schema_storage_path: str | Path | None = None,
    **_: Any,
) -> tuple[Self | None, list[ValidationErrorType]]:
    """Read a table instance from file.

    Args:
        instance_path: The path to the Schema definition YAML file.
        catalog_name: The name of the Catalog of the Table.
        schema_name: The name of the Schema of the Table.
        schema_storage_path: The storage path location of the Schema of the Table.

    Returns:
        - Instance of the class that was successfully instantiated and
              validated
        - The second list contains errors encountered during the process,
              which could be validation errors or YAML parsing/scanning errors.

    Raises:
        FileNotFoundError: If the specified directory does not exist or is
            not a directory and fail_on_missing_subfolder is
            True.
        ValueError: If catalog_name or schema_name are not provided.
    """
    processed_instance_path = process_path(instance_path)
    if not processed_instance_path:
        raise FileNotFoundError("Table file not found.")
    schema_storage_path = process_path(schema_storage_path)
    errors: list[ValidationErrorType] = []

    if not catalog_name or not schema_name:
        errors.append(ValueError("catalog_name and schema_name must be provided."))
        return None, errors

    try:
        with processed_instance_path.open("r") as file:
            data = yaml.safe_load(file)
            data["identifier"] = f"{catalog_name}.{schema_name}.{data['name']}"
            if data.get("is_external"):
                if storage_path := data.get("storage_path"):
                    data["storage_path"] = storage_path
                elif schema_storage_path:
                    data["storage_path"] = (schema_storage_path / data["name"]).as_posix()
                else:
                    raise ValueError(
                        f"Neither storage path nor schema storage path of table {data['name']} has been provided."
                    )

            instance, sub_errors = cls.metadata_to_instance(data)
            errors += sub_errors
    except (
        ValidationError,
        ParserError,
        ScannerError,
    ) as e:
        instance = None
        errors.append(e)
    return instance, errors

read_instances_from_directory(instance_path, fail_on_missing_subfolder=True, catalog_name=None, schema_name=None, schema_storage_path=None, **_) classmethod

Reads instances from a directory containing YAML files.

This method scans a specified directory for YAML files (.yaml or .yml), attempts to read and parse each file as an instance of the class, and collects any errors encountered during the process. If the directory does not exist or is not a directory, it either raises a FileNotFoundError (if fail_on_missing_subfolder is True) or returns empty lists.

Parameters:

Name Type Description Default
instance_path str | Path

The path to the directory containing instance files.

 required
catalog_name str | None

Name of the catalog to which these instances belong.

 None
schema_name str | None

Name of the schema used for validating the instances.

 None
fail_on_missing_subfolder bool

Determines behavior when the specified directory does not exist or is not a directory. Defaults to True, which will raise a FileNotFoundError.

 True
schema_storage_path str | Path | None

Path to the storage location of the schema these tables instances belong to.

 None

Returns:

Type Description
list[Self]
  • The first list contains instances of the class that were successfully read and validated from the files.
list[ValidationErrorType]
  • The second list contains errors encountered during the process, which could be validation errors or YAML parsing/scanning errors.

Raises:

Type Description
FileNotFoundError

If the specified directory does not exist or is not a directory and fail_on_missing_subfolder is True.
ValueError

If catalog_name or schema_name are not provided.
Source code in src/cloe_nessy/models/table.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@classmethod
def read_instances_from_directory(
    cls,
    instance_path: str | Path,
    fail_on_missing_subfolder: bool = True,
    catalog_name: str | None = None,
    schema_name: str | None = None,
    schema_storage_path: str | Path | None = None,
    **_: Any,
) -> tuple[list[Self], list[ValidationErrorType]]:
    """Reads instances from a directory containing YAML files.

    This method scans a specified directory for YAML files (.yaml or .yml),
    attempts to read and parse each file as an instance of the class, and
    collects any errors encountered during the process. If the directory
    does not exist or is not a directory, it either raises a
    FileNotFoundError (if fail_on_missing_subfolder is True) or returns
    empty lists.

    Args:
        instance_path: The path to the directory containing instance files.
        catalog_name: Name of the catalog to which these instances belong.
        schema_name: Name of the schema used for validating the instances.
        fail_on_missing_subfolder: Determines behavior when the specified
                directory does not exist or is not a directory. Defaults to True,
                which will raise a FileNotFoundError.
        schema_storage_path: Path to the storage location of the schema
                these tables instances belong to.

    Returns:
        - The first list contains instances of the class that were
            successfully read and validated from the files.
        - The second list contains errors encountered during the
            process, which could be validation errors or YAML
            parsing/scanning errors.

    Raises:
        FileNotFoundError: If the specified directory does not exist or is
                           not a directory and fail_on_missing_subfolder is True.
        ValueError: If catalog_name or schema_name are not provided.
    """
    processed_instance_path = process_path(instance_path)
    schema_storage_path = process_path(schema_storage_path)
    errors: list[ValidationErrorType] = []

    if not catalog_name or not schema_name:
        errors.append(ValueError("catalog_name and schema_name must be provided."))
        return [], errors
    instances: list[Self] = []

    if not processed_instance_path or not processed_instance_path.exists() or not processed_instance_path.is_dir():
        if fail_on_missing_subfolder:
            raise FileNotFoundError(f"Directory not found: {processed_instance_path}")
        else:
            return instances, errors

    for instance_file in processed_instance_path.iterdir():
        sub_errors: list[ValidationErrorType] = []
        if instance_file.is_file() and instance_file.suffix in (".yaml", ".yml"):
            instance, sub_errors = cls.read_instance_from_file(
                instance_file, catalog_name, schema_name, str(schema_storage_path)
            )
            instances += [] if instance is None else [instance]
        errors += sub_errors

    return instances, errors

remove_column(column)

Remove a column from the Table.

Args. column: The column to be removed.

Source code in src/cloe_nessy/models/table.py 
293
294
295
296
297
298
299
300
301
302
303
304
def remove_column(self, column: str | Column) -> None:
    """Remove a column from the Table.

    Args.
        column: The column to be removed.
    """
    if isinstance(column, Column):
        column_name = column.name
    else:
        column_name = column

    self.columns = [col for col in self.columns if col.name != column_name]

update_column(column)

Replaces a Column with a new Column object to update it.

Parameters:

Name Type Description Default
column Column

The new column object, to replace the old one.

 required
Source code in src/cloe_nessy/models/table.py 
276
277
278
279
280
281
282
283
def update_column(self, column: Column) -> None:
    """Replaces a Column with a new Column object to update it.

    Args:
        column: The new column object, to replace the old one.
    """
    self.remove_column(column)
    self.add_column(column)

Volume

Bases: TemplateLoaderMixin, LoggerMixin, BaseModel

Volume class for managing volumes.

Source code in src/cloe_nessy/models/volume.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class Volume(TemplateLoaderMixin, LoggerMixin, BaseModel):
    """Volume class for managing volumes."""

    identifier: str
    storage_path: str | Path
    comment: str | None = None

    @field_validator("identifier")
    def check_identifier(cls, value):
        """Check the identifier."""
        if value.count(".") != 2:
            raise ValueError("The identifier must be in the format 'catalog.schema.volume_name'.")
        return value

    @property
    def storage_identifier(self) -> str:
        """Return the storage identifier."""
        return f"/Volumes/{self.catalog}/{self.schema}/{self.name}/"

    @property
    def catalog(self) -> str:
        """Return the catalog name."""
        return self.identifier.split(".")[0]

    @property
    def schema_name(self) -> str:
        """Return the schema name."""
        return self.identifier.split(".")[1]

    @property
    def name(self) -> str:
        """Return the table name."""
        return self.identifier.split(".")[2]

    @property
    def escaped_identifier(self) -> str:
        """Return the escaped identifier."""
        return f"`{self.catalog}`.`{self.schema_name}`.`{self.name}`"

    def model_post_init(self, __context: Any) -> None:
        """Post init method for the Table model."""
        self._console_logger = self.get_console_logger()
        self._console_logger.debug(f"Model for volume [ '{self.identifier}' ] has been initialized.")

    def get_create_statement(
        self,
        if_not_exists: bool = True,
    ):
        """Get the create statement for the Volume.

        Args:
            if_not_exists: Whether to include the IF NOT EXISTS clause in the create statement

        Returns:
                The rendered create statement as a string.
        """
        template_name: str = "create_volume.sql.j2"
        templates = Path(__file__).parent / "templates"

        try:
            template = self.get_template(templates, template_name)
        except TemplateNotFound as err:
            self._console_logger.error(f"Template [ {template_name} ] not found.")
            raise err
        render = template.render(volume=self, if_not_exists=if_not_exists)
        return render

catalog property

Return the catalog name.

escaped_identifier property

Return the escaped identifier.

name property

Return the table name.

schema_name property

Return the schema name.

storage_identifier property

Return the storage identifier.

check_identifier(value)

Check the identifier.

Source code in src/cloe_nessy/models/volume.py 
18
19
20
21
22
23
@field_validator("identifier")
def check_identifier(cls, value):
    """Check the identifier."""
    if value.count(".") != 2:
        raise ValueError("The identifier must be in the format 'catalog.schema.volume_name'.")
    return value

get_create_statement(if_not_exists=True)

Get the create statement for the Volume.

Parameters:

Name Type Description Default
if_not_exists bool

Whether to include the IF NOT EXISTS clause in the create statement

 True

Returns:

Type Description

The rendered create statement as a string.
Source code in src/cloe_nessy/models/volume.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def get_create_statement(
    self,
    if_not_exists: bool = True,
):
    """Get the create statement for the Volume.

    Args:
        if_not_exists: Whether to include the IF NOT EXISTS clause in the create statement

    Returns:
            The rendered create statement as a string.
    """
    template_name: str = "create_volume.sql.j2"
    templates = Path(__file__).parent / "templates"

    try:
        template = self.get_template(templates, template_name)
    except TemplateNotFound as err:
        self._console_logger.error(f"Template [ {template_name} ] not found.")
        raise err
    render = template.render(volume=self, if_not_exists=if_not_exists)
    return render

model_post_init(__context)

Post init method for the Table model.

Source code in src/cloe_nessy/models/volume.py 
50
51
52
53
def model_post_init(self, __context: Any) -> None:
    """Post init method for the Table model."""
    self._console_logger = self.get_console_logger()
    self._console_logger.debug(f"Model for volume [ '{self.identifier}' ] has been initialized.")