Skip to content

OpenSematicLab Controller

A controller extents a model class with functions

Entity

Bases: Entity

Source code in src/osw/controller/entity.py 
4
5
6
class Entity(model.Entity):
    def explain(self):
        print(f"Entity with label '{str(self.label[0].text)}'")

DatabaseController

Bases: Database

Provides a sqlalchemy engine for the specified database

Source code in src/osw/controller/database.py 
 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
 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
class DatabaseController(model.Database):
    """
    Provides a sqlalchemy engine for the specified database
    """

    osw: Optional[OSW]
    """ an OSW instance to fetch related resources (host, server, etc.)"""
    cm: Optional[Union[CredentialManager.BaseCredential, CredentialManager]]
    """ CredentialManager to login to the database"""
    engine: Optional[Engine]
    """ the internal sqlalchemy engine """

    class Config:
        arbitrary_types_allowed = True

    class ConnectionString(OswBaseModel):
        """the sqlalchemy database connection string"""

        dialect: str
        """ sql subtype, e. g. postgres"""
        username: str
        """ username for the login """
        password: str
        """ password for the login """
        host: str
        """ the database host (ip address or domain) """
        port: int
        """ the database port """
        database: str
        """ the database name / identifier """
        driver: Optional[str]
        """ specific driver, e. g. psycopg2 """

        def __str__(self):
            """generates the string representation

            Returns
            -------
                database connection string
            """
            prefix = f"{self.dialect}"
            if self.driver:
                prefix += f"+{self.driver}"
            url = URL.create(
                prefix,
                username=self.username,
                password=self.password,
                host=self.host,
                port=self.port,
                database=self.database,
            )
            return url.render_as_string(hide_password=False)

    class ConnectionConfig(OswBaseModel):
        """Database connection configuration"""

        osw: OSW
        """ OSW instance to fetch related resources (host, server, etc.) """
        cm: Union[CredentialManager.BaseCredential, CredentialManager]
        """ CredentialManager or direct Credential to login to the database"""

    def connect(self, config: ConnectionConfig):
        """Initializes the connection to the database by creating a sqlalchemy engine

        Parameters
        ----------
        config
            see ConnectionConfig
        """
        self.osw = config.osw
        self.cm = config.cm

        server_title = self.osw.site.semantic_search(
            f"[[-HasDbServer::Item:{self.osw.get_osw_id(self.uuid)}]]"
        )
        server = self.osw.load_entity(server_title[0]).cast(model.DatabaseServer)

        host_title = self.osw.site.semantic_search(
            f"[[-HasHost::Item:{self.osw.get_osw_id(server.uuid)}]]"
        )
        host = self.osw.load_entity(host_title[0]).cast(model.Host)

        dbtype_title = self.osw.site.semantic_search(
            f"[[-HasDbType::Item:{self.osw.get_osw_id(server.uuid)}]]"
        )
        dbtype = self.osw.load_entity(dbtype_title[0]).cast(model.DatabaseType)

        if isinstance(self.cm, CredentialManager.BaseCredential):
            db_server_cred = self.cm
        else:
            db_server_cred = config.cm.get_credential(
                CredentialManager.CredentialConfig(
                    iri=f"{host.network_domain[0]}:{server.network_port[0]}",
                    fallback=CredentialManager.CredentialFallback.ask,
                )
            )

        cstr = DatabaseController.ConnectionString(
            dialect=dbtype.connection_str_dialect,
            driver=dbtype.connection_str_driver,
            username=db_server_cred.username,
            password=db_server_cred.password,
            host=host.network_domain[0],
            port=server.network_port[0],
            database=self.name,
        )

        self.engine = create_engine(str(cstr))

    def execute(self, sql: str):
        """Executes a plain sql string

        Parameters
        ----------
        sql
            the sql string
        """
        with self.engine.connect() as conn:
            result_set = conn.execute(sql_text(sql))
            for r in result_set:
                print(r)

cm instance-attribute

CredentialManager to login to the database

engine instance-attribute

the internal sqlalchemy engine

osw instance-attribute

an OSW instance to fetch related resources (host, server, etc.)

ConnectionConfig dataclass

Bases: OswBaseModel

Database connection configuration

Source code in src/osw/controller/database.py 
66
67
68
69
70
71
72
class ConnectionConfig(OswBaseModel):
    """Database connection configuration"""

    osw: OSW
    """ OSW instance to fetch related resources (host, server, etc.) """
    cm: Union[CredentialManager.BaseCredential, CredentialManager]
    """ CredentialManager or direct Credential to login to the database"""

cm instance-attribute

CredentialManager or direct Credential to login to the database

osw instance-attribute

OSW instance to fetch related resources (host, server, etc.)

ConnectionString dataclass

Bases: OswBaseModel

the sqlalchemy database connection string

Source code in src/osw/controller/database.py 
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
class ConnectionString(OswBaseModel):
    """the sqlalchemy database connection string"""

    dialect: str
    """ sql subtype, e. g. postgres"""
    username: str
    """ username for the login """
    password: str
    """ password for the login """
    host: str
    """ the database host (ip address or domain) """
    port: int
    """ the database port """
    database: str
    """ the database name / identifier """
    driver: Optional[str]
    """ specific driver, e. g. psycopg2 """

    def __str__(self):
        """generates the string representation

        Returns
        -------
            database connection string
        """
        prefix = f"{self.dialect}"
        if self.driver:
            prefix += f"+{self.driver}"
        url = URL.create(
            prefix,
            username=self.username,
            password=self.password,
            host=self.host,
            port=self.port,
            database=self.database,
        )
        return url.render_as_string(hide_password=False)

database instance-attribute

the database name / identifier

dialect instance-attribute

sql subtype, e. g. postgres

driver instance-attribute

specific driver, e. g. psycopg2

host instance-attribute

the database host (ip address or domain)

password instance-attribute

password for the login

port instance-attribute

the database port

username instance-attribute

username for the login

__str__()

generates the string representation

Returns:

Type Description
database connection string
Source code in src/osw/controller/database.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __str__(self):
    """generates the string representation

    Returns
    -------
        database connection string
    """
    prefix = f"{self.dialect}"
    if self.driver:
        prefix += f"+{self.driver}"
    url = URL.create(
        prefix,
        username=self.username,
        password=self.password,
        host=self.host,
        port=self.port,
        database=self.database,
    )
    return url.render_as_string(hide_password=False)

connect(config)

Initializes the connection to the database by creating a sqlalchemy engine

Parameters:

Name Type Description Default
config ConnectionConfig

see ConnectionConfig

 required
Source code in src/osw/controller/database.py 
 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
def connect(self, config: ConnectionConfig):
    """Initializes the connection to the database by creating a sqlalchemy engine

    Parameters
    ----------
    config
        see ConnectionConfig
    """
    self.osw = config.osw
    self.cm = config.cm

    server_title = self.osw.site.semantic_search(
        f"[[-HasDbServer::Item:{self.osw.get_osw_id(self.uuid)}]]"
    )
    server = self.osw.load_entity(server_title[0]).cast(model.DatabaseServer)

    host_title = self.osw.site.semantic_search(
        f"[[-HasHost::Item:{self.osw.get_osw_id(server.uuid)}]]"
    )
    host = self.osw.load_entity(host_title[0]).cast(model.Host)

    dbtype_title = self.osw.site.semantic_search(
        f"[[-HasDbType::Item:{self.osw.get_osw_id(server.uuid)}]]"
    )
    dbtype = self.osw.load_entity(dbtype_title[0]).cast(model.DatabaseType)

    if isinstance(self.cm, CredentialManager.BaseCredential):
        db_server_cred = self.cm
    else:
        db_server_cred = config.cm.get_credential(
            CredentialManager.CredentialConfig(
                iri=f"{host.network_domain[0]}:{server.network_port[0]}",
                fallback=CredentialManager.CredentialFallback.ask,
            )
        )

    cstr = DatabaseController.ConnectionString(
        dialect=dbtype.connection_str_dialect,
        driver=dbtype.connection_str_driver,
        username=db_server_cred.username,
        password=db_server_cred.password,
        host=host.network_domain[0],
        port=server.network_port[0],
        database=self.name,
    )

    self.engine = create_engine(str(cstr))

execute(sql)

Executes a plain sql string

Parameters:

Name Type Description Default
sql str

the sql string

 required
Source code in src/osw/controller/database.py 
122
123
124
125
126
127
128
129
130
131
132
133
def execute(self, sql: str):
    """Executes a plain sql string

    Parameters
    ----------
    sql
        the sql string
    """
    with self.engine.connect() as conn:
        result_set = conn.execute(sql_text(sql))
        for r in result_set:
            print(r)