Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
import asyncio
from handofcats import as_command
from minidb import AsyncTable
class Users(AsyncTable):
pk = "name"
users = Users([{"name": "foo"}, {"name": "bar"}, {"name": "boo"}])
async def run_queries():
async def do_task(name):
print(await users.find_one(in_=[name]))
await asyncio.gather(*[do_task(name) for name in ["foo", "bar", "boo"]])
async def run_bulk_query():
bulk_fut = asyncio.Future()
input_buf = []
async def find_user(name: str):
input_buf.append(name)
for row in await bulk_fut:
if row["name"] == name:
return row
async def do_bulk(n):
await asyncio.sleep(n)
bulk_fut.set_result(await users.find_all(in_=input_buf))
async def do_task(name):
print(await find_user(name))
actions = [do_task(name) for name in ["foo", "bar", "boo"]]
actions.append(do_bulk(0.1))
await asyncio.gather(*actions)
@as_command
def run():
asyncio.run(run_queries())
asyncio.run(run_bulk_query())
import minidb
import asyncio
class Users(minidb.AsyncTable):
pk = "name"
users = Users([{"name": "foo"}, {"name": "bar"}, {"name": "boo"}])
async def run():
return await users.find_all(in_=["foo", "boo"])
got = asyncio.run(run())
print(got)
# [{'name': 'foo'}, {'name': 'boo'}]
import minidb
import asyncio
class Users(minidb.AsyncTable):
pk = "name"
users = Users([{"name": "foo"}, {"name": "bar"}, {"name": "boo"}])
async def run():
print(await users.find_one(in_=["foo"]))
# {'name': 'foo'}
print(await users.find_one(in_=["boo"]))
# {'name': 'boo'}
asyncio.run(run())
import logging
import time
import typing as t
logger = logging.getLogger(__name__)
Doc = t.Dict[str, t.Any]
PK = str
class Table:
pk: t.ClassVar[PK] = "id"
def __init__(self, data: t.List[Doc]) -> None:
self.data = data
def find_all(
self,
*,
where: t.Optional[t.Callable[[Doc], bool]] = None,
in_: t.Optional[t.List[PK]] = None,
) -> t.List[Doc]:
logger.info("%s find_all", self.__class__.__name__)
time.sleep(0.1)
data = self.data
if in_ is not None:
data = [d for d in data if d[self.pk] in in_]
if where is not None:
data = [d for d in data if where(d)]
return data
def find_one(
self,
*,
where: t.Optional[t.Callable[[Doc], bool]] = None,
in_: t.Optional[t.List[PK]] = None,
) -> t.List[Doc]:
logger.info("%s find_one", self.__class__.__name__)
time.sleep(0.1)
data = self.data
if in_ is not None:
data = [d for d in data if d[self.pk] in in_]
if where is not None:
data = [d for d in data if where(d)]
return data[0]
class AsyncTable(Table):
def find_all(
self,
*,
where: t.Optional[t.Callable[[Doc], bool]] = None,
in_: t.Optional[t.List[PK]] = None,
) -> t.Awaitable[t.List[Doc]]:
import asyncio
from functools import partial
return asyncio.get_event_loop().run_in_executor(
None, partial(super().find_all, where=where, in_=in_)
)
def find_one(
self,
*,
where: t.Optional[t.Callable[[Doc], bool]] = None,
in_: t.Optional[t.List[PK]] = None,
) -> t.List[Doc]:
import asyncio
from functools import partial
return asyncio.get_event_loop().run_in_executor(
None, partial(super().find_one, where=where, in_=in_)
)
if __name__ == "__main__":
import unittest
class Users(Table):
pass
foo = {"id": "1", "name": "foo"}
bar = {"id": "2", "name": "bar"}
boo = {"id": "3", "name": "boo"}
data = [foo, bar, boo]
users = Users(data)
class Tests(unittest.TestCase):
def test_find_all(self):
got = users.find_all()
self.assertEqual(got, data)
def test_find_all__in(self):
got = users.find_all(in_=["2"])
self.assertEqual(got, [bar])
def test_find_all__where(self):
got = users.find_all(where=lambda d: d["name"].endswith("oo"))
self.assertEqual(got, [foo, boo])
unittest.main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment