Skip to content

Instantly share code, notes, and snippets.

@banteg
Created July 12, 2017 04:02
Show Gist options
  • Save banteg/9848f1edbc0612c788ea7cdc3f78139c to your computer and use it in GitHub Desktop.
Save banteg/9848f1edbc0612c788ea7cdc3f78139c to your computer and use it in GitHub Desktop.
import ujson
import asyncpg
class DB:
def __init__(self):
self.pool = None
async def init(self, *args, **kwds):
self.pool = await asyncpg.create_pool(*args, init=self.codecs, **kwds)
def __getattr__(self, item):
return getattr(self.pool, item)
@staticmethod
async def codecs(conn: asyncpg.Connection):
await conn.set_type_codec('json', encoder=ujson.dumps, decoder=ujson.loads, schema='pg_catalog', format='text')
db = DB()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment