schema_sentinel.metadata_manager.model.function

 1import json
 2
 3import sqlalchemy as db
 4from sqlalchemy import ForeignKey, select
 5
 6from . import CommonBase
 7
 8
 9class Function(CommonBase):
10    __tablename__ = "functions"
11    function_id = db.Column(db.String, primary_key=True)
12    argument_signature = db.Column(db.String, primary_key=True)
13    schema_id = db.Column(db.String, ForeignKey("schemas.schema_id"))
14    function_name = db.Column(db.String)
15    function_owner = db.Column(db.String)
16    data_type = db.Column(db.String)
17    character_maximum_length = db.Column(db.Integer)
18    character_octet_length = db.Column(db.Integer)
19    numeric_precision = db.Column(db.Integer)
20    numeric_precision_radix = db.Column(db.Integer)
21    numeric_scale = db.Column(db.Integer)
22    datetime_scale = db.Column(db.Integer)
23    function_language = db.Column(db.String)
24    function_definition = db.Column(db.String)
25    volatility = db.Column(db.String)
26    is_null_call = db.Column(db.String)
27    is_secure = db.Column(db.String)
28    comment = db.Column(db.String)
29    created = db.Column(db.String)
30    last_altered = db.Column(db.String)
31    is_external = db.Column(db.String)
32    api_integration = db.Column(db.String)
33    context_headers = db.Column(db.String)
34    max_batch_rows = db.Column(db.Integer)
35    compression = db.Column(db.String)
36    packages = db.Column(db.String)
37    runtime_version = db.Column(db.String)
38    installed_packages = db.Column(db.String)
39    is_memoizable = db.Column(db.String)
40
41    def save(self, session) -> None:
42        if not session.execute(self.exists()).first():
43            session.add(self)
44            session.commit()
45
46    def exists(self) -> str:
47        return select(Function).filter_by(function_id=self.function_id)
48
49    def __get_id__(self) -> str:
50        id = json.loads(self.schema_id)
51        id["function_name"] = self.function_name
52        id["argument_signature"] = self.argument_signature
53        return json.dumps(id)
54
55    def __data_type__(self) -> str:
56        if self.data_type in ["VARCHAR", "TEXT"]:
57            data_type = f"{self.data_type}({self.character_maximum_length})"
58        elif self.data_type == "NUMBER":
59            data_type = f"{self.data_type}({self.numeric_precision}, {self.numeric_scale})"
60        elif self.data_type in ["TIMESTAMP_NTZ", "TIMESTAMP_LTZ", "TIMESTAMP_TZ"]:
61            data_type = f"{self.data_type}({self.datetime_presision})"
62        else:
63            data_type = self.data_type
64        return data_type