schema_sentinel.metadata_manager.model.column
1import json 2 3import sqlalchemy as db 4from sqlalchemy import ForeignKey, select 5 6from . import CommonBase 7 8 9class Column(CommonBase): 10 __tablename__ = "columns" 11 column_id = db.Column(db.String, primary_key=True) 12 table_id = db.Column(db.String, ForeignKey("tables.table_id")) 13 column_name = db.Column(db.String) 14 ordinal_position = db.Column(db.Integer) 15 column_default = db.Column(db.String) 16 is_nullable = db.Column(db.String, default="YES") 17 data_type = db.Column(db.String) 18 character_maximum_length = db.Column(db.Integer) 19 character_octet_length = db.Column(db.Integer) 20 numeric_precision = db.Column(db.Integer) 21 numeric_precision_radix = db.Column(db.Integer) 22 numeric_scale = db.Column(db.Integer) 23 datetime_precision = db.Column(db.Integer) 24 is_identity = db.Column(db.String, default="NO") 25 identity_generation = db.Column(db.String) 26 identity_start = db.Column(db.String) 27 identity_increment = db.Column(db.String) 28 comment = db.Column(db.String) 29 30 def save(self, session) -> None: 31 if not session.execute(self.exists()).first(): 32 session.add(self) 33 session.commit() 34 35 def exists(self) -> str: 36 return select(Column).filter_by(column_id=self.column_id) 37 38 def __get_id__(self) -> str: 39 id = json.loads(self.table_id) 40 id["column_name"] = self.column_name 41 return json.dumps(id) 42 43 def __data_type__(self) -> str: 44 if self.data_type in ["VARCHAR", "TEXT"]: 45 data_type = f"{self.data_type}({self.character_maximum_length})" 46 elif self.data_type == "NUMBER": 47 data_type = f"{self.data_type}({self.numeric_precision}, {self.numeric_scale})" 48 elif self.data_type in ["TIMESTAMP_NTZ", "TIMESTAMP_LTZ", "TIMESTAMP_TZ"]: 49 data_type = f"{self.data_type}({self.datetime_presision})" 50 else: 51 data_type = self.data_type 52 return data_type
10class Column(CommonBase): 11 __tablename__ = "columns" 12 column_id = db.Column(db.String, primary_key=True) 13 table_id = db.Column(db.String, ForeignKey("tables.table_id")) 14 column_name = db.Column(db.String) 15 ordinal_position = db.Column(db.Integer) 16 column_default = db.Column(db.String) 17 is_nullable = db.Column(db.String, default="YES") 18 data_type = db.Column(db.String) 19 character_maximum_length = db.Column(db.Integer) 20 character_octet_length = db.Column(db.Integer) 21 numeric_precision = db.Column(db.Integer) 22 numeric_precision_radix = db.Column(db.Integer) 23 numeric_scale = db.Column(db.Integer) 24 datetime_precision = db.Column(db.Integer) 25 is_identity = db.Column(db.String, default="NO") 26 identity_generation = db.Column(db.String) 27 identity_start = db.Column(db.String) 28 identity_increment = db.Column(db.String) 29 comment = db.Column(db.String) 30 31 def save(self, session) -> None: 32 if not session.execute(self.exists()).first(): 33 session.add(self) 34 session.commit() 35 36 def exists(self) -> str: 37 return select(Column).filter_by(column_id=self.column_id) 38 39 def __get_id__(self) -> str: 40 id = json.loads(self.table_id) 41 id["column_name"] = self.column_name 42 return json.dumps(id) 43 44 def __data_type__(self) -> str: 45 if self.data_type in ["VARCHAR", "TEXT"]: 46 data_type = f"{self.data_type}({self.character_maximum_length})" 47 elif self.data_type == "NUMBER": 48 data_type = f"{self.data_type}({self.numeric_precision}, {self.numeric_scale})" 49 elif self.data_type in ["TIMESTAMP_NTZ", "TIMESTAMP_LTZ", "TIMESTAMP_TZ"]: 50 data_type = f"{self.data_type}({self.datetime_presision})" 51 else: 52 data_type = self.data_type 53 return data_type
The base class of the class hierarchy.
When called, it accepts no arguments and returns a new featureless instance that has no instance attributes and cannot be given any.
Column(**kwargs)
A simple constructor that allows initialization from kwargs.
Sets attributes on the constructed instance using the names and
values in kwargs.
Only keys that are present as attributes of the instance's class are allowed. These could be, for example, any mapped columns or relationships.