Skip to content

volume

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_name}/{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,
    ) -> str:
        """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 cast(str, 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
str

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,
) -> str:
    """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 cast(str, 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.")