aiophoenixdb 0.0.6

Asyncio with phoenixdb

What is aiophoenixdb

This project is based on the Apache Software Foundation open source project Apache-Phoenixdb project to transform the call implementation of the Avatica protocol in the code from the original synchronous mode to asynchronous call.

Getting started

How to install

pip install aiophoenixdb

How to use

• Query sample

import aiophoenixdb
import asyncio

PHOENIX_CONFIG = {
    'url': 'http://xxxxxxxxxx',
    'user': 'xxx',
    'password': 'xxx',
    'database': 'xxx'
}

async def query_test():
    conn = await aiophoenixdb.connect(**PHOENIX_CONFIG)
    async with conn:
        async with conn.cursor() as ps:
            # need await
            await ps.execute("SELECT * FROM xxx WHERE id = ?",  parameters=("1", ))
            res = await ps.fetchone()
            print(res)

# Throw the query coroutine into the event loop to run
asyncio.get_event_loop().run_until_complete(query_test())

• Query with DictCursor

import aiophoenixdb
import asyncio
from aiophoenixdb.cursors import DictCursor

PHOENIX_CONFIG = {
    'url': 'http://xxxxxxxxxx',
    'user': 'xxx',
    'password': 'xxx',
    'database': 'xxx'
}

async def query_test():
    conn = await aiophoenixdb.connect(**PHOENIX_CONFIG)
    async with conn:
        async with conn.cursor(cursor_factory=DictCursor) as ps:
            # need await
            await ps.execute("SELECT * FROM xxx WHERE id = ?",  parameters=("1", ))
            res = await ps.fetchone()
            print(res)

# Throw the query coroutine into the event loop to run
asyncio.get_event_loop().run_until_complete(query_test())