schema_sentinel.metadata_manager.model.database 

 1import json
 2
 3import sqlalchemy as db
 4from sqlalchemy import select
 5
 6from . import CommonBase
 7
 8
 9class Database(CommonBase):
10    __tablename__ = "databases"
11    database_id = db.Column(db.String, primary_key=True)
12    version = db.Column(db.String)
13    environment = db.Column(db.String)
14    database_name = db.Column(db.String)
15    database_owner = db.Column(db.String)
16    is_transient = db.Column(db.String, default="NO")
17    comment = db.Column(db.String)
18    created = db.Column(db.String)
19    last_altered = db.Column(db.String)
20    retention_time = db.Column(db.String)
21
22    def save(self, session) -> None:
23        if not session.execute(self.exists()).first():
24            session.add(self)
25            session.commit()
26
27    def exists(self) -> str:
28        return select(Database).filter_by(database_id=self.database_id)
29
30    def __get_id__(self) -> str:
31        return json.dumps(
32            {"database_name": self.database_name, "version": self.version, "environment": self.environment}
33        )
34
35    def __get_name__(self) -> str:
36        return f"{self.database_name} v.{self.version}({self.environment})"
class Database(schema_sentinel.metadata_manager.model.CommonBase): 
10class Database(CommonBase):
11    __tablename__ = "databases"
12    database_id = db.Column(db.String, primary_key=True)
13    version = db.Column(db.String)
14    environment = db.Column(db.String)
15    database_name = db.Column(db.String)
16    database_owner = db.Column(db.String)
17    is_transient = db.Column(db.String, default="NO")
18    comment = db.Column(db.String)
19    created = db.Column(db.String)
20    last_altered = db.Column(db.String)
21    retention_time = db.Column(db.String)
22
23    def save(self, session) -> None:
24        if not session.execute(self.exists()).first():
25            session.add(self)
26            session.commit()
27
28    def exists(self) -> str:
29        return select(Database).filter_by(database_id=self.database_id)
30
31    def __get_id__(self) -> str:
32        return json.dumps(
33            {"database_name": self.database_name, "version": self.version, "environment": self.environment}
34        )
35
36    def __get_name__(self) -> str:
37        return f"{self.database_name} v.{self.version}({self.environment})"

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.

Database(**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.

database_id
version
environment
database_name
database_owner
is_transient
comment
created
last_altered
retention_time
def save(self, session) -> None: 
23    def save(self, session) -> None:
24        if not session.execute(self.exists()).first():
25            session.add(self)
26            session.commit()
def exists(self) -> str: 
28    def exists(self) -> str:
29        return select(Database).filter_by(database_id=self.database_id)