pbx/app0/cdr_test.py

#!/usr/bin/env python3
import asyncio
import aiohttp
import logging
import os
import re
import json
import sys
from datetime import datetime as dt
from datetime import timedelta as td
from typing import Any, Optional
from functools import wraps
from secrets import compare_digest
from databases import Database
from cel import *
from logging.config import dictConfig
from pprint import pformat
from inspect import getmembers

_db = Database('mysql://{}:{}@{}:{}/{}'.format(os.getenv('MYSQL_USER', 'asterisk'),
                                               os.getenv('MYSQL_PASSWORD', 'secret'),
                                               os.getenv('MYSQL_SERVER', 'db'),
                                               os.getenv('APP_PORT_MYSQL', '3306'),
                                               os.getenv('FREEPBX_CDRDBNAME', None)))
linkedid = sys.argv[1]

async def main():
  await _db.connect()
  _q = '''SELECT *
          FROM cdr
          WHERE linkedid=:linkedid
          ORDER BY sequence;'''
  _v = {'linkedid': linkedid}
  _f = True
  async for row in _db.iterate(query=_q, values=_v):
    print('\t'.join([str(k) for k in row]))
  await _db.disconnect()

if __name__ ==  '__main__':
  loop = asyncio.get_event_loop()
  loop.run_until_complete(main())