simple_oracle.py

simple_oracle.py

class MinhaClasseBonita:
    #...
    
    @classmethod
    async def get_one(cls, oracle_pool, **kargs ):
        retval = {}
        async with oracle_pool.acquire() as connection:
            async with connection.cursor() as cursor:
                # retorna dicts
                sql_stmt = cls.get_sql()
                assert sql_stmt.startswith('SELECT '), 'SQL não é um "SELECT"'
                await cursor.execute(
                    sql_stmt,
                    **kargs
                )
                cursor._cursor.rowfactory = lambda *arguments: dict(zip([d[0].lower() for d in cursor._cursor.description], arguments))
                record = await cursor.fetchone()
                retval = cls(**record)
                return retval
            
    @classmethod
    async def get_one_closing_cursor(cls, oracle_pool, **kargs ):
        retval = {}
        async with oracle_pool.acquire() as connection:
            cursor = await connection.cursor()
            sql_stmt = cls.get_sql()
            assert sql_stmt.startswith('SELECT '), 'SQL não é um "SELECT"'
            await cursor.execute(
                sql_stmt,
                **kargs
            )
            cursor._cursor.rowfactory = lambda *arguments: dict(zip([d[0].lower() for d in cursor._cursor.description], arguments))
            record = await cursor.fetchone()
            # MAGIC
            cursor._cursor.close()
            retval = cls(**record)
            return retval

    @classmethod
    def sync_get_one(cls, oracle_pool, **kargs ):
        retval = {}
        # Acquire a connection from the pool
        connection = oracle_pool.acquire()

        # Use the pooled connection
        cursor = connection.cursor()
        sql_stmt = cls.get_sql()
        assert sql_stmt.startswith('SELECT '), 'SQL não é um "SELECT"'
        cursor.execute(
            sql_stmt,
            **kargs
        )
        cursor.rowfactory = lambda *arguments: dict(zip([d[0].lower() for d in cursor.description], arguments))
        record = cursor.fetchone()
        cursor.close()
        retval = cls(**record)

        # Release the connection to the pool
        oracle_pool.release(connection)
        return retval