Skip to content

Model

API Status

API documentation is auto-generated from docstrings. Ensure docstrings are comprehensive.

Overview

[Module purpose and key concepts]

Reference

model

This module provides the class Model.

The class Model is responsible for the connection to the PostgreSQL database.

Once connected, you can use the get_relation_class <#half_orm.model.Model.get_relation_class>_ method to generate a class to access any relation (table/view) in your database.

Example

from half_orm.model import Model model = Model('my_config_file') class MyTable(model.get_relation_class('my_schema.my_table')): ... # Your business code goes here

Note

The default schema is public in PostgreSQL, so to reference a table my_table in this schema you'll have to use pubic.my_table.

Classes

Model

Parameters:

Name Type Description Default
config_file str

the configuration file that contains the informations to connect to the database.

 required
scope Optional[str]

used to agregate several modules in a package. See hop <https://github.com/collorg/halfORM/blob/main/doc/hop.md>_.

 None
Note

The config_file is searched in the HALFORM_CONF_DIR variable if specified, then in /etc/half_orm. The file format is as follows:

| [database]
| name = <postgres db name>
| user = <postgres user>
| password = <postgres password>
| host = <host name | localhost>
| port = <port | 5432>

name is the only mandatory entry if you are using an ident login with a local account <https://www.postgresql.org/docs/current/auth-ident.html>_.

Source code in half_orm/model.py 
 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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
class Model:
    """
    Parameters:
        config_file (str): the configuration file that contains the informations to connect
            to the database.
        scope (Optional[str]): used to agregate several modules in a package.
            See `hop <https://github.com/collorg/halfORM/blob/main/doc/hop.md>`_.

    Note:
        The **config_file** is searched in the `HALFORM_CONF_DIR` variable if specified,
        then in `/etc/half_orm`. The file format is as follows:

            | [database]
            | name = <postgres db name>
            | user = <postgres user>
            | password = <postgres password>
            | host = <host name | localhost>
            | port = <port | 5432>

        *name* is the only mandatory entry if you are using an
        `ident login with a local account <https://www.postgresql.org/docs/current/auth-ident.html>`_.
    """
    __deja_vu = {}
    _classes_ = {}
    def __init__(self, config_file: None, scope: str=None):
        """Model constructor

        Use @config_file in your scripts. The @dbname parameter is
        reserved to the __factory metaclass.
        """
        self.__dbinfo = {}
        self.__load_config(config_file)
        self._scope = scope and scope.split('.')[0]
        self.__conn = None
        self.__connect()

    def __load_config(self, config_file):
        """Load the config file

        Raises:
            MissingConfigFile: If the **config_file** is not found in *HALFORM_CONF_DIR*.
            MalformedConfigFile: if the *name* is missing in the **config_file**.
            RuntimeError: If the reconnection is attempted on another database.
        """
        self.__config_file = config_file
        config = ConfigParser()
        file_ = os.path.join(CONF_DIR, self.__config_file)
        if config.read([file_]):
            try:
                database = config['database']
            except KeyError as exc:
                raise model_errors.MalformedConfigFile(file_, 'Missing section', 'database') from exc
            try:
                dbname = database['name']
            except KeyError as exc:
                raise model_errors.MalformedConfigFile(file_, 'Missing mandatory parameter', 'name') from exc

            if self.__dbinfo and dbname != self.__dbname:
                raise RuntimeError(
                    f"Can't reconnect to another database: {dbname} != {self.__dbname}")
            self.__dbinfo['dbname'] = dbname

        else:
            dbname = config_file
            self.__dbinfo['dbname'] = dbname
            database = {'user': None, 'password': None, 'host': None, 'port': None}
            if dbname != 'template1':
                utils.warning(f"No config file '{os.path.join(CONF_DIR, dbname)}'.\n\t Trying peer authentication for '{os.environ.get('USER') or os.getlogin()}'.\n")

        self.__dbinfo['user'] = database.get('user')
        self.__dbinfo['password'] = database.get('password')
        self.__dbinfo['host'] = database.get('host')
        self.__dbinfo['port'] = database.get('port')
        self.__dbinfo['connect_timeout'] = database.get('timeout', 3)

    def __connect(self, config_file: str=None, reload: bool=False):
        """Setup a new connection to the database.

        The reconnect method is an alias to the ``__connect`` method.

        Parameters:
            config_file (str): If a config_file is provided, the connection is made with the new
                parameters, allowing to change role. The database name must be the same.
            reload (bool): If set to True, reloads the metadata from the database. Usefull if
                the model has changed.
        """
        self.disconnect()

        if config_file:
            self.__load_config(config_file)
        self.__conn = psycopg2.connect(**self.__dbinfo, cursor_factory=RealDictCursor)
        self.__conn.autocommit = True
        self.__pg_meta = pg_meta.PgMeta(self.__conn, reload)
        if reload:
            self._classes_[self._dbname] = {}
        if self.__dbname not in self.__class__.__deja_vu:
            self.__deja_vu[self.__dbname] = self

    reconnect = __connect

    def get_relation_class(self, relation_name: str, fields_aliases: typing.Dict=None): # -> Relation
        """This method is a factory that generates a class that inherits the `Relation <#half_orm.relation.Relation>`_ class.

        Args:
            relation_name (string): the full name (`<schema>.<relation>`) of the targeted relation in the database.

        Raises:
            ValueError: if the schema is missing in relation_name
            UnknownRelationError: if the relation is not found in the database

        Returns:
            a class that inherits the `Relation <#half_orm.relation.Relation>`_ class:
                the class corresponding to the relation in the database.

        Examples:
            A class inheriting the `Relation <#half_orm.relation.Relation>`_ class is returned:
                >>> Person = model.get_relation_class('actor.person')
                >>> Person
                <class 'half_orm.relation.Table_HalftestActorPerson'>
                >>> Person.__bases__
                (<class 'half_orm.relation.Relation'>,)

            A prefered way to create a class:
                >>> class Person(model.get_relation_class('actor.person)):
                >>>     # Your code goes here

            A `MissingSchemaInName <#half_orm.model_errors.MissingSchemaInName>`_ is raised when the schema name is missing:
                >>> model.get_relation_class('person')
                [...]MissingSchemaInName: do you mean 'public.person'?

            An `UnknownRelation <#half_orm.model_errors.UnknownRelation>`_ is raised if the relation is not found in the model:
                >>> model.get_relation_class('public.person')
                [...]UnknownRelation: 'public.person' does not exist in the database halftest.
        """
        try:
            schema, table = relation_name.replace('"', '').rsplit('.', 1)
        except ValueError as err:
            raise model_errors.MissingSchemaInName(relation_name) from err
        return factory({'fqrn': (self.__dbname, schema, table), 'model': self.__deja_vu[self.__dbname], 'fields_aliases':fields_aliases})


    @staticmethod
    def _deja_vu(dbname):
        """Returns None if the database hasn't been loaded yet.
        Otherwise, it returns the Model object already loaded.
        The Model object is shared between all_ the relations in the
        database. The Model object is loaded only once for a given database.
        """
        return Model.__deja_vu.get(dbname)

    @property
    def __dbname(self):
        return self.__dbinfo['dbname']

    def ping(self):
        """Checks if the connection is still established.
        Attempts a new connection otherwise.

        Returns:
            bool: True if the connection is established, False otherwise.
        """
        try:
            self.execute_query("select 1")
            return True
        except (psycopg2.OperationalError, psycopg2.InterfaceError):
            try:
                self.__connect()
                self.execute_query("select 1")
            except (psycopg2.OperationalError, psycopg2.InterfaceError) as exc: #pragma: no cover
                # log reconnection attempt failure
                sys.stderr.write(f'{exc.exception}\n')
                sys.stderr.flush()
            return False

    def disconnect(self):
        """Closes the connection to the database.
        """
        if self.__conn is not None and not self.__conn.closed:
            self.__conn.close()

    def _reload(self, config_file=None):
        """Reload metadata

        Updates the model according to changes made to the database.
        """
        self.__connect(config_file, True)

    @property
    def _dbname(self):
        """
        property. Returns the database name.
        """
        return self.__dbname

    @property
    def _connection(self):
        """\
        Property. Returns the psycopg2 connection attached to the Model object.
        """
        return self.__conn

    def _fields_metadata(self, sfqrn):
        "Proxy to PgMeta.fields_meta"
        return self.__pg_meta.fields_meta(self.__dbname, sfqrn)

    def _fkeys_metadata(self, sfqrn):
        "Proxy to PgMeta.fkeys_meta"
        return self.__pg_meta.fkeys_meta(self.__dbname, sfqrn)

    def _relation_metadata(self, fqrn):
        "Proxy to PgMeta.relation_meta"
        return self.__pg_meta.relation_meta(self.__dbname, fqrn)

    def _unique_constraints_list(self, fqrn):
        "Proxy to PgMeta._unique_constraints_list"
        return self.__pg_meta._unique_constraints_list(self.__dbname, fqrn)

    def _pkey_constraint(self, fqrn):
        "Proxy to PgMeta._pkey_constraint"
        return self.__pg_meta._pkey_constraint(self.__dbname, fqrn)

    def execute_query(self, query, values=(), mogrify=False):
        """Executes a raw SQL query.

        Warning:
            This method calls the psycopg2
            `cursor.execute <https://www.psycopg.org/docs/cursor.html?highlight=execute#cursor.execute>`_
            function.
            Please read the psycopg2 documentation on
            `passing parameters to SQL queries <https://www.psycopg.org/docs/usage.html#query-parameters>`_.
        """
        cursor = self.__conn.cursor(cursor_factory=RealDictCursor)
        try:
            if mogrify:
                print(cursor.mogrify(query, values).decode('utf-8'))
            cursor.execute(query, values)
        except (psycopg2.OperationalError, psycopg2.InterfaceError):
            self.ping()
            cursor = self.__conn.cursor(cursor_factory=RealDictCursor)
            cursor.execute(query, values)
        return cursor

    def execute_function(self, fct_name, *args, **kwargs) -> typing.List[tuple]:
        """`Executes a PostgreSQL function <https://www.postgresql.org/docs/current/sql-syntax-calling-funcs.html>`_.

        Arguments:
            *args: The list of parameters to be passed to the postgresql function.
            **kwargs: The list of named parameters to be passed to the postgresql function.

        Returns:
            List[tuple]: a list of tuples.

        Raises:
            RuntimeError: If you mix ***args** and ****kwargs**.

        Note:
            You can't mix args and kwargs with the execute_function method!
        """
        if bool(args) and bool(kwargs):
            raise RuntimeError("You can't mix args and kwargs with the execute_function method!")
        cursor = self.__conn.cursor(cursor_factory=RealDictCursor)
        if kwargs:
            values = kwargs
        else:
            values = args
        cursor.callproc(fct_name, values)
        return cursor.fetchall()

    def call_procedure(self, proc_name, *args, **kwargs):
        """`Executes a PostgreSQL procedure <https://www.postgresql.org/docs/current/sql-call.html>`_.

        Arguments:
            *args: The list of parameters to be passed to the postgresql function.
            **kwargs: The list of named parameters to be passed to the postgresql function.

        Returns:
            None | List[tuple]: None or a list of tuples.

        Raises:
            RuntimeError: If you mix ***args** and ****kwargs**.

        Note:
            You can't mix args and kwargs with the call_procedure method!
        """
        if bool(args) and bool(kwargs):
            raise RuntimeError("You can't mix args and kwargs with the call_procedure method!")
        if kwargs:
            params = ', '.join([f'{key} => %s' for key in kwargs])
            values = tuple(kwargs.values())
        else:
            params = ', '.join(['%s' for _ in range(len(args))])
            values = args
        query = f'call {proc_name}({params})'
        cursor = self.__conn.cursor(cursor_factory=RealDictCursor)
        cursor.execute(query, values)
        try:
            return cursor.fetchall()
        except psycopg2.ProgrammingError:
            return None

    def has_relation(self, qtn: str) -> bool:
        """Checks if the qtn is a relation in the database

        Returns:
            bool: True if the relation exists in the database, False otherwise.

        Example:
            >>> model.has_relation('public.person')
            False
            >>> model.has_relation('actor.person')
            True
        """
        return self.__pg_meta.has_relation(self.__dbname, *qtn.rsplit('.', 1))

    def _import_class(self, qtn, scope=None):
        """Used to return the class from the scope module.

        This method is used to import a class from a module. The module
        must reside in an accessible python package named `scope`.
        """
        t_qtn = qtn.replace('"', '').rsplit('.', 1)
        self._scope = scope or self._scope
        module_path = ".".join(t_qtn)
        if self._scope:
            module_path = f'{self._scope}.{module_path}'
        _class_name = pg_meta.class_name(qtn) # XXX
        try:
            module = __import__(
                module_path, globals(), locals(), [_class_name], 0)
            return module.__dict__[_class_name]
        except:
            return self.get_relation_class(qtn)

    def _relations(self):
        """List all_ the relations in the database"""
        rels = self.__pg_meta.relations_list(self.__dbname)
        return rels

    def desc(self):
        """Returns the list of the relations of the model.

        Each element in the list contains:

        * the relation type: 'r' relation, 'v' view, 'm' materialized view, 'p' partition;
        * a tuple identifying the relation: (db name>, <schema name>, <relation name>);
        * a list of tuples indentifying the inherited relations.

        Example:
            >>> from half_orm.model import Model
            >>> halftest = Model('halftest')
            >>> halftest.desc()
            [('r', ('halftest', 'actor', 'person'), []), ('r', ('halftest', 'blog', 'comment'), []), ('r', ('halftest', 'blog', 'event'), [('halftest', 'blog', 'post')]), ('r', ('halftest', 'blog', 'post'), []), ('v', ('halftest', 'blog.view', 'post_comment'), [])]
        """
        return self.__pg_meta.desc(self.__dbname)

    def __str__(self):
        return self.__pg_meta.str(self.__dbname)

    def classes(self):
        "Returns the all the classes of the model"
        for relation in self._relations():
            package_name = relation[1][0]
            module_name = ".".join(relation[1][1:])
            if module_name.find('half_orm_meta') == 0:
                continue
            class_name = pg_meta.camel_case(relation[1][-1])
            module = importlib.import_module(f".{module_name}", package_name)
            yield getattr(module, class_name), relation[0]
Functions
__connect(config_file=None, reload=False)

Setup a new connection to the database.

The reconnect method is an alias to the __connect method.

Parameters:

Name Type Description Default
config_file str

If a config_file is provided, the connection is made with the new parameters, allowing to change role. The database name must be the same.

 None
reload bool

If set to True, reloads the metadata from the database. Usefull if the model has changed.

 False
Source code in half_orm/model.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def __connect(self, config_file: str=None, reload: bool=False):
    """Setup a new connection to the database.

    The reconnect method is an alias to the ``__connect`` method.

    Parameters:
        config_file (str): If a config_file is provided, the connection is made with the new
            parameters, allowing to change role. The database name must be the same.
        reload (bool): If set to True, reloads the metadata from the database. Usefull if
            the model has changed.
    """
    self.disconnect()

    if config_file:
        self.__load_config(config_file)
    self.__conn = psycopg2.connect(**self.__dbinfo, cursor_factory=RealDictCursor)
    self.__conn.autocommit = True
    self.__pg_meta = pg_meta.PgMeta(self.__conn, reload)
    if reload:
        self._classes_[self._dbname] = {}
    if self.__dbname not in self.__class__.__deja_vu:
        self.__deja_vu[self.__dbname] = self
__init__(config_file, scope=None)

Model constructor

Use @config_file in your scripts. The @dbname parameter is reserved to the __factory metaclass.

Source code in half_orm/model.py 
69
70
71
72
73
74
75
76
77
78
79
def __init__(self, config_file: None, scope: str=None):
    """Model constructor

    Use @config_file in your scripts. The @dbname parameter is
    reserved to the __factory metaclass.
    """
    self.__dbinfo = {}
    self.__load_config(config_file)
    self._scope = scope and scope.split('.')[0]
    self.__conn = None
    self.__connect()
__load_config(config_file)

Load the config file

Raises:

Type Description
MissingConfigFile

If the config_file is not found in HALFORM_CONF_DIR.
MalformedConfigFile

if the name is missing in the config_file.
RuntimeError

If the reconnection is attempted on another database.
Source code in half_orm/model.py 
 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
def __load_config(self, config_file):
    """Load the config file

    Raises:
        MissingConfigFile: If the **config_file** is not found in *HALFORM_CONF_DIR*.
        MalformedConfigFile: if the *name* is missing in the **config_file**.
        RuntimeError: If the reconnection is attempted on another database.
    """
    self.__config_file = config_file
    config = ConfigParser()
    file_ = os.path.join(CONF_DIR, self.__config_file)
    if config.read([file_]):
        try:
            database = config['database']
        except KeyError as exc:
            raise model_errors.MalformedConfigFile(file_, 'Missing section', 'database') from exc
        try:
            dbname = database['name']
        except KeyError as exc:
            raise model_errors.MalformedConfigFile(file_, 'Missing mandatory parameter', 'name') from exc

        if self.__dbinfo and dbname != self.__dbname:
            raise RuntimeError(
                f"Can't reconnect to another database: {dbname} != {self.__dbname}")
        self.__dbinfo['dbname'] = dbname

    else:
        dbname = config_file
        self.__dbinfo['dbname'] = dbname
        database = {'user': None, 'password': None, 'host': None, 'port': None}
        if dbname != 'template1':
            utils.warning(f"No config file '{os.path.join(CONF_DIR, dbname)}'.\n\t Trying peer authentication for '{os.environ.get('USER') or os.getlogin()}'.\n")

    self.__dbinfo['user'] = database.get('user')
    self.__dbinfo['password'] = database.get('password')
    self.__dbinfo['host'] = database.get('host')
    self.__dbinfo['port'] = database.get('port')
    self.__dbinfo['connect_timeout'] = database.get('timeout', 3)
call_procedure(proc_name, *args, **kwargs)

Executes a PostgreSQL procedure <https://www.postgresql.org/docs/current/sql-call.html>_.

Parameters:

Name Type Description Default
*args

The list of parameters to be passed to the postgresql function.

 ()
**kwargs

The list of named parameters to be passed to the postgresql function.

 {}

Returns:

Type Description

None | List[tuple]: None or a list of tuples.

Raises:

Type Description
RuntimeError

If you mix ***args** and ****kwargs**.
Note

You can't mix args and kwargs with the call_procedure method!

Source code in half_orm/model.py 
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
def call_procedure(self, proc_name, *args, **kwargs):
    """`Executes a PostgreSQL procedure <https://www.postgresql.org/docs/current/sql-call.html>`_.

    Arguments:
        *args: The list of parameters to be passed to the postgresql function.
        **kwargs: The list of named parameters to be passed to the postgresql function.

    Returns:
        None | List[tuple]: None or a list of tuples.

    Raises:
        RuntimeError: If you mix ***args** and ****kwargs**.

    Note:
        You can't mix args and kwargs with the call_procedure method!
    """
    if bool(args) and bool(kwargs):
        raise RuntimeError("You can't mix args and kwargs with the call_procedure method!")
    if kwargs:
        params = ', '.join([f'{key} => %s' for key in kwargs])
        values = tuple(kwargs.values())
    else:
        params = ', '.join(['%s' for _ in range(len(args))])
        values = args
    query = f'call {proc_name}({params})'
    cursor = self.__conn.cursor(cursor_factory=RealDictCursor)
    cursor.execute(query, values)
    try:
        return cursor.fetchall()
    except psycopg2.ProgrammingError:
        return None
classes()

Returns the all the classes of the model

Source code in half_orm/model.py 
403
404
405
406
407
408
409
410
411
412
def classes(self):
    "Returns the all the classes of the model"
    for relation in self._relations():
        package_name = relation[1][0]
        module_name = ".".join(relation[1][1:])
        if module_name.find('half_orm_meta') == 0:
            continue
        class_name = pg_meta.camel_case(relation[1][-1])
        module = importlib.import_module(f".{module_name}", package_name)
        yield getattr(module, class_name), relation[0]
desc()

Returns the list of the relations of the model.

Each element in the list contains:

  • the relation type: 'r' relation, 'v' view, 'm' materialized view, 'p' partition;
  • a tuple identifying the relation: (db name>, , );
  • a list of tuples indentifying the inherited relations.
Example

from half_orm.model import Model halftest = Model('halftest') halftest.desc() [('r', ('halftest', 'actor', 'person'), []), ('r', ('halftest', 'blog', 'comment'), []), ('r', ('halftest', 'blog', 'event'), [('halftest', 'blog', 'post')]), ('r', ('halftest', 'blog', 'post'), []), ('v', ('halftest', 'blog.view', 'post_comment'), [])]

Source code in half_orm/model.py 
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
def desc(self):
    """Returns the list of the relations of the model.

    Each element in the list contains:

    * the relation type: 'r' relation, 'v' view, 'm' materialized view, 'p' partition;
    * a tuple identifying the relation: (db name>, <schema name>, <relation name>);
    * a list of tuples indentifying the inherited relations.

    Example:
        >>> from half_orm.model import Model
        >>> halftest = Model('halftest')
        >>> halftest.desc()
        [('r', ('halftest', 'actor', 'person'), []), ('r', ('halftest', 'blog', 'comment'), []), ('r', ('halftest', 'blog', 'event'), [('halftest', 'blog', 'post')]), ('r', ('halftest', 'blog', 'post'), []), ('v', ('halftest', 'blog.view', 'post_comment'), [])]
    """
    return self.__pg_meta.desc(self.__dbname)
disconnect()

Closes the connection to the database.

Source code in half_orm/model.py 
219
220
221
222
223
def disconnect(self):
    """Closes the connection to the database.
    """
    if self.__conn is not None and not self.__conn.closed:
        self.__conn.close()
execute_function(fct_name, *args, **kwargs)

Executes a PostgreSQL function <https://www.postgresql.org/docs/current/sql-syntax-calling-funcs.html>_.

Parameters:

Name Type Description Default
*args

The list of parameters to be passed to the postgresql function.

 ()
**kwargs

The list of named parameters to be passed to the postgresql function.

 {}

Returns:

Type Description
List[tuple]

List[tuple]: a list of tuples.

Raises:

Type Description
RuntimeError

If you mix ***args** and ****kwargs**.
Note

You can't mix args and kwargs with the execute_function method!

Source code in half_orm/model.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def execute_function(self, fct_name, *args, **kwargs) -> typing.List[tuple]:
    """`Executes a PostgreSQL function <https://www.postgresql.org/docs/current/sql-syntax-calling-funcs.html>`_.

    Arguments:
        *args: The list of parameters to be passed to the postgresql function.
        **kwargs: The list of named parameters to be passed to the postgresql function.

    Returns:
        List[tuple]: a list of tuples.

    Raises:
        RuntimeError: If you mix ***args** and ****kwargs**.

    Note:
        You can't mix args and kwargs with the execute_function method!
    """
    if bool(args) and bool(kwargs):
        raise RuntimeError("You can't mix args and kwargs with the execute_function method!")
    cursor = self.__conn.cursor(cursor_factory=RealDictCursor)
    if kwargs:
        values = kwargs
    else:
        values = args
    cursor.callproc(fct_name, values)
    return cursor.fetchall()
execute_query(query, values=(), mogrify=False)

Executes a raw SQL query.

Warning

This method calls the psycopg2 cursor.execute <https://www.psycopg.org/docs/cursor.html?highlight=execute#cursor.execute>_ function. Please read the psycopg2 documentation on passing parameters to SQL queries <https://www.psycopg.org/docs/usage.html#query-parameters>_.

Source code in half_orm/model.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
def execute_query(self, query, values=(), mogrify=False):
    """Executes a raw SQL query.

    Warning:
        This method calls the psycopg2
        `cursor.execute <https://www.psycopg.org/docs/cursor.html?highlight=execute#cursor.execute>`_
        function.
        Please read the psycopg2 documentation on
        `passing parameters to SQL queries <https://www.psycopg.org/docs/usage.html#query-parameters>`_.
    """
    cursor = self.__conn.cursor(cursor_factory=RealDictCursor)
    try:
        if mogrify:
            print(cursor.mogrify(query, values).decode('utf-8'))
        cursor.execute(query, values)
    except (psycopg2.OperationalError, psycopg2.InterfaceError):
        self.ping()
        cursor = self.__conn.cursor(cursor_factory=RealDictCursor)
        cursor.execute(query, values)
    return cursor
get_relation_class(relation_name, fields_aliases=None)

This method is a factory that generates a class that inherits the Relation <#half_orm.relation.Relation>_ class.

Parameters:

Name Type Description Default
relation_name string

the full name (<schema>.<relation>) of the targeted relation in the database.

 required

Raises:

Type Description
ValueError

if the schema is missing in relation_name
UnknownRelationError

if the relation is not found in the database

Returns:

Type Description

a class that inherits the Relation <#half_orm.relation.Relation>_ class: the class corresponding to the relation in the database.

Examples:

A class inheriting the Relation <#half_orm.relation.Relation>_ class is returned: >>> Person = model.get_relation_class('actor.person') >>> Person >>> Person.bases (,)

A prefered way to create a class: >>> class Person(model.get_relation_class('actor.person)): >>> # Your code goes here

A MissingSchemaInName <#half_orm.model_errors.MissingSchemaInName>_ is raised when the schema name is missing: >>> model.get_relation_class('person') [...]MissingSchemaInName: do you mean 'public.person'?

An UnknownRelation <#half_orm.model_errors.UnknownRelation>_ is raised if the relation is not found in the model: >>> model.get_relation_class('public.person') [...]UnknownRelation: 'public.person' does not exist in the database halftest.

Source code in half_orm/model.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def get_relation_class(self, relation_name: str, fields_aliases: typing.Dict=None): # -> Relation
    """This method is a factory that generates a class that inherits the `Relation <#half_orm.relation.Relation>`_ class.

    Args:
        relation_name (string): the full name (`<schema>.<relation>`) of the targeted relation in the database.

    Raises:
        ValueError: if the schema is missing in relation_name
        UnknownRelationError: if the relation is not found in the database

    Returns:
        a class that inherits the `Relation <#half_orm.relation.Relation>`_ class:
            the class corresponding to the relation in the database.

    Examples:
        A class inheriting the `Relation <#half_orm.relation.Relation>`_ class is returned:
            >>> Person = model.get_relation_class('actor.person')
            >>> Person
            <class 'half_orm.relation.Table_HalftestActorPerson'>
            >>> Person.__bases__
            (<class 'half_orm.relation.Relation'>,)

        A prefered way to create a class:
            >>> class Person(model.get_relation_class('actor.person)):
            >>>     # Your code goes here

        A `MissingSchemaInName <#half_orm.model_errors.MissingSchemaInName>`_ is raised when the schema name is missing:
            >>> model.get_relation_class('person')
            [...]MissingSchemaInName: do you mean 'public.person'?

        An `UnknownRelation <#half_orm.model_errors.UnknownRelation>`_ is raised if the relation is not found in the model:
            >>> model.get_relation_class('public.person')
            [...]UnknownRelation: 'public.person' does not exist in the database halftest.
    """
    try:
        schema, table = relation_name.replace('"', '').rsplit('.', 1)
    except ValueError as err:
        raise model_errors.MissingSchemaInName(relation_name) from err
    return factory({'fqrn': (self.__dbname, schema, table), 'model': self.__deja_vu[self.__dbname], 'fields_aliases':fields_aliases})
has_relation(qtn)

Checks if the qtn is a relation in the database

Returns:

Name Type Description
bool bool

True if the relation exists in the database, False otherwise.
Example

model.has_relation('public.person') False model.has_relation('actor.person') True

Source code in half_orm/model.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
def has_relation(self, qtn: str) -> bool:
    """Checks if the qtn is a relation in the database

    Returns:
        bool: True if the relation exists in the database, False otherwise.

    Example:
        >>> model.has_relation('public.person')
        False
        >>> model.has_relation('actor.person')
        True
    """
    return self.__pg_meta.has_relation(self.__dbname, *qtn.rsplit('.', 1))
ping()

Checks if the connection is still established. Attempts a new connection otherwise.

Returns:

Name Type Description
bool

True if the connection is established, False otherwise.
Source code in half_orm/model.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def ping(self):
    """Checks if the connection is still established.
    Attempts a new connection otherwise.

    Returns:
        bool: True if the connection is established, False otherwise.
    """
    try:
        self.execute_query("select 1")
        return True
    except (psycopg2.OperationalError, psycopg2.InterfaceError):
        try:
            self.__connect()
            self.execute_query("select 1")
        except (psycopg2.OperationalError, psycopg2.InterfaceError) as exc: #pragma: no cover
            # log reconnection attempt failure
            sys.stderr.write(f'{exc.exception}\n')
            sys.stderr.flush()
        return False

Functions