aioradio

Generic asynchronous i/o python utilities for AWS services (SQS, S3, DynamoDB, Secrets Manager), Redis, MSSQL (pyodbc), JIRA and more.

AWS S3 example code

aioradio abstracts using aiobotocore and aioboto3 making async AWS funtion calls simple one liners. Besides what is shown below in the examples, there is also support for SQS, DynamoDB and Secrets Manager.

import asyncio

from aioradio.aws.s3 import (
    create_bucket,
    delete_s3_object,
    download_file,
    get_object,
    list_s3_objects,
    upload_file
)

async def main():
    s3_bucket = 'aioradio'
    s3_prefix = 'test'
    filename = 'hello_world.txt'
    s3_key = f'{s3_prefix}/{filename}'

    # create an s3 bucket called aioradio
    await create_bucket(bucket=s3_bucket)

    # create hello_world.txt file
    with open(filename, 'w') as file_handle:
        file_handle.write('hello world of aioradio!')

    # upload the file from s3 and confirm it now exists in s3
    await upload_file(bucket=s3_bucket, filepath=filename, s3_key=s3_key)
    assert s3_key in await list_s3_objects(bucket=s3_bucket, s3_prefix=s3_prefix)

    # test downloading the file
    await download_file(bucket=s3_bucket, filepath=filename, s3_key=s3_key)

    # test getting file data to object
    result = await get_object(bucket=s3_bucket, s3_key=s3_key)
    assert result == b'hello world of aioradio!'

    # delete the file from s3
    await delete_s3_object(bucket=s3_bucket, s3_prefix=s3_key)
    assert s3_key not in await list_s3_objects(bucket=s3_bucket, s3_prefix=s3_prefix)

asyncio.get_event_loop().run_until_complete(main())

MSSQL example code

aioredis uses the pyodbc library to work with ODBC databases. It currently has support for connecting and sending queries to mssql.

import asyncio

from aioradio.pyodbc import establish_pyodbc_connection
from aioradio.pyodbc import pyodbc_query_fetchone
from aioradio.pyodbc import pyodbc_query_fetchall

def main():
    conn = establish_pyodbc_connection(host='your-host', user='your-user', pwd='your-password')

    query = "SELECT homeruns FROM MLB.dbo.LosAngelesAngels WHERE lastname = 'Trout' AND year = '2020'"
    row = pyodbc_query_fetchone(conn=conn, query=query)
    print(row)

    query = "SELECT homeruns FROM MLB.dbo.LosAngelesAngels WHERE lastname = 'Trout'"
    rows = pyodbc_query_fetchall(conn=conn, query=query)
    print(rows)


asyncio.get_event_loop().run_until_complete(main())

Jira example code

Jira uses the async library httpx behind the scene to send http requests.

import asyncio

from aioradio.jira import add_comment_to_jira
from aioradio.jira import get_jira_issue
from aioradio.jira import post_jira_issue

async def main():

    # create a jira ticket
    url = 'https://aioradio.atlassian.net/rest/api/2/issue/'
    payload = {
        "fields": {
            "project": {"key": "aioradio"},
            "issuetype": {"name": "Task"},
            "reporter": {"accountId": "somebodies-account-id"},
            "priority": {"name": "Medium"},
            "summary": "Aioradio rocks!",
            "description": "Aioradio Review",
            "labels": ["aioradio"],
            "assignee": {"accountId": "somebodies-account-id"}
        }
    }
    resp = await post_jira_issue(url=url, jira_user='your-user', jira_token='your-password', payload=payload)
    jira_id = resp.json()['key']

    # get jira ticket info
    resp = await get_jira_issue(url=f'{url}/{jira_id}', jira_user='your-user', jira_token='your-password')

    # add comment to jira ticket
    comment = 'aioradio rocks!'
    response = await add_comment_to_jira(url=url, jira_user='your-user', jira_token='your-password', comment=comment)

asyncio.get_event_loop().run_until_complete(main())

INSTALLING FOR DIRECT DEVELOPMENT OF AIORADIO

Install python 3.10.X

Make sure you've installed ODBC drivers, required for using the python package pyodbc.

Clone aioradio locally and navigate to the root directory

Install and activate python VirtualEnv

python3.10 -m venv env
source env/bin/activate

Install python modules included in requirements.txt

pip install -r aioradio/requirements.txt

Run Makefile command from the root directory to test all is good before issuing push to master

make all

AUTHORS

Tim Reichard - aioradio

See also the list of contributors who participated in this project.

ACKNOWLEDGEMENTS

Pedro Artiga - Developer contributing to aioradio.

Release files for aioradio 0.17.11

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for aioradio 0.17.11
File	Size	Uploaded
aioradio-0.17.11.tar.gz	35.1 kB	Aug 17, 2022	Details

Built distribution (wheel)

Table of built distributions (wheels) for aioradio 0.17.11
File	Interpreter	ABI	Platform	Reset
aioradio-0.17.11-py3-none-any.whl 38.8 kB Aug 17, 2022	Python 3	none	any	Details

Total release size: 74.0 kB

Release files / aioradio-0.17.11.tar.gz

Download URL	aioradio-0.17.11.tar.gz
Size	35.1 kB
Tags	Source
SHA-256 checksum How to use checksums	`8a315d4ef5e5db97bca4e1ca997c40085019096b5bde2c39dbf18b0a1cd32dad`
BLAKE2b-256 checksum How to use checksums	`4c11c606242490ce365fb240a736819fe812c36c9c0de5eebad1d35e6ef13006`
Upload date	Aug 17, 2022
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/4.0.1 CPython/3.10.4`

Release files / aioradio-0.17.11-py3-none-any.whl

Download URL	aioradio-0.17.11-py3-none-any.whl
Size	38.8 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`598501c7ecd5464e4b9418244cc36878bcb0fcbd07fbbb71e74b08d9a1c1f41e`
BLAKE2b-256 checksum How to use checksums	`dabc24b13c04fbb75108156ae599fbe662e3955cac22161243552bf7b74c56d8`
Upload date	Aug 17, 2022
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/4.0.1 CPython/3.10.4`