Module src.knxrcore.sql.mysql

Classes

class Connection (**kw)

Expand source code

class Connection:
    def __init__(self, **kw) -> None:
        self.user = kw.get('user', 'root')
        self.password = kw.get('password', '')
        self.host = kw.get('host', 'localhost')
        self.db = kw.get('db', '')

    def get(self, sql: str, params: Any = '') -> list[dict | tuple] | None:
        """ Here we can select something from the Database.

        Parameters
        ----------
        sql : :class:`str`
            An sql string to select something from the Database.

        params : :class:`Any`
            The parameters for the sql execution.

        Returns
        -------
        list | List[Optional[dict]] | List[Tuple]
            Returns the result of the given query.
        """

        conn = connect(user=self.user, password=self.password, host=self.host, database=self.db)
        cur = conn.cursor(buffered=True)

        cur.execute(sql, params)
        result = cur.fetchall()

        conn.commit()
        cur.close()
        return result

    def update(self, sql: str, params: Any = '') -> Any:
        """ With this method you can Update the Database, like `UPDATE` or `DELETE` query's

        Parameters
        ----------
        sql : :class:`str`
            An sql string to select something from the Database.

        params : :class:`Any`
            The parameters for the sql execution.

        Returns
        -------
        None
            Nothing.
        """

        return self.get(sql, params)

    async def aget(self, sql: str, params: Any = '') -> list[dict | tuple | None]:
        """ Here we can select asynchronous something from the Database.

        Parameters
        ----------
        sql : :class:`str`
            An sql string to select something from the Database.

        params : :class:`Any`
            The parameters for the sql execution.

        Returns
        -------
        list | List[Optional[dict]] | List[Tuple]
            Returns the result of the given query.
        """

        return self.get(sql, params)

    async def aupdate(self, sql: str, params: Any = '') -> Any:
        """ With this method you can Update the Database, such things like `UPDATE` or `DELETE` query's

       Parameters
       ----------
       sql : :class:`str`
           An sql string to select something from the Database.

       params : :class:`Any`
           The parameters for the sql execution.

       Returns
       -------
       None
           Nothing.
       """

        return self.update(sql, params)

Methods

async def aget(self, sql: str, params: Any = '') ‑> list[dict | tuple | None]

Here we can select asynchronous something from the Database.

Parameters

sql : :class:str``

An sql string to select something from the Database.

params : :class:Any``

The parameters for the sql execution.

Returns

list | List[Optional[dict]] | List[Tuple]

Returns the result of the given query.

Expand source code

async def aget(self, sql: str, params: Any = '') -> list[dict | tuple | None]:
    """ Here we can select asynchronous something from the Database.

    Parameters
    ----------
    sql : :class:`str`
        An sql string to select something from the Database.

    params : :class:`Any`
        The parameters for the sql execution.

    Returns
    -------
    list | List[Optional[dict]] | List[Tuple]
        Returns the result of the given query.
    """

    return self.get(sql, params)

async def aupdate(self, sql: str, params: Any = '') ‑> Any

With this method you can Update the Database, such things like UPDATE or DELETE query's

Parameters

sql : :class:str``

An sql string to select something from the Database.

params : :class:Any``

The parameters for the sql execution.

Returns

None

Nothing.

Expand source code

async def aupdate(self, sql: str, params: Any = '') -> Any:
    """ With this method you can Update the Database, such things like `UPDATE` or `DELETE` query's

   Parameters
   ----------
   sql : :class:`str`
       An sql string to select something from the Database.

   params : :class:`Any`
       The parameters for the sql execution.

   Returns
   -------
   None
       Nothing.
   """

    return self.update(sql, params)

def get(self, sql: str, params: Any = '') ‑> list[dict | tuple] | None

Here we can select something from the Database.

Parameters

sql : :class:str``

An sql string to select something from the Database.

params : :class:Any``

The parameters for the sql execution.

Returns

list | List[Optional[dict]] | List[Tuple]

Returns the result of the given query.

Expand source code

def get(self, sql: str, params: Any = '') -> list[dict | tuple] | None:
    """ Here we can select something from the Database.

    Parameters
    ----------
    sql : :class:`str`
        An sql string to select something from the Database.

    params : :class:`Any`
        The parameters for the sql execution.

    Returns
    -------
    list | List[Optional[dict]] | List[Tuple]
        Returns the result of the given query.
    """

    conn = connect(user=self.user, password=self.password, host=self.host, database=self.db)
    cur = conn.cursor(buffered=True)

    cur.execute(sql, params)
    result = cur.fetchall()

    conn.commit()
    cur.close()
    return result

def update(self, sql: str, params: Any = '') ‑> Any

With this method you can Update the Database, like UPDATE or DELETE query's

Parameters

sql : :class:str``

An sql string to select something from the Database.

params : :class:Any``

The parameters for the sql execution.

Returns

None

Nothing.

Expand source code

def update(self, sql: str, params: Any = '') -> Any:
    """ With this method you can Update the Database, like `UPDATE` or `DELETE` query's

    Parameters
    ----------
    sql : :class:`str`
        An sql string to select something from the Database.

    params : :class:`Any`
        The parameters for the sql execution.

    Returns
    -------
    None
        Nothing.
    """

    return self.get(sql, params)