Task Group demo code.

mikeckennedy · mikeckennedy · commit cd80bf01db16 · 2022-12-20T16:24:22.000-08:00
diff --git a/code/ch03-concurrency/groups.py b/code/ch03-concurrency/groups.py
@@ -0,0 +1,75 @@
+import asyncio
+from asyncio import TaskGroup
+from pprint import pprint
+
+import httpx
+from typing import List, Optional
+
+import pydantic
+from pydantic import BaseModel
+
+
+class SearchItem(BaseModel):
+    category: str
+    id: Optional[int]
+    item_id: Optional[int]
+    url: str
+    title: str
+    description: str
+
+
+class SearchResponse(BaseModel):
+    elapsed_ms: float
+    keywords: List[str]
+    results: List[SearchItem] = pydantic.Field(default_factory=list)
+    episodes: List[SearchItem] = pydantic.Field(default_factory=list)
+
+
+def cleanup(resp: SearchResponse):
+    # Unfortunately, Talk Python's and Python Bytes' APIs vary slightly.
+    # This method will unify them.
+    resp.episodes = resp.episodes or resp.results
+    resp.results = resp.episodes or resp.results
+
+    for r in resp.results:
+        r.id = r.id or r.item_id
+
+
+async def search_podcast(text: str, server: str) -> list[str]:
+    search_url = f'https://{server}/api/search?q={text}'
+    async with httpx.AsyncClient() as client:
+        resp = await client.get(search_url)
+        resp.raise_for_status()
+
+    data = resp.json()
+    resp = SearchResponse(**data)
+    cleanup(resp)
+
+    return [f'{r.id}: {r.title}' for r in resp.results if r.category.lower() == 'episode']
+
+
+async def search():
+    # pydantic beanie mongodb odm <-- is a good search string
+    text = input("What keyword to you want to look for? ").strip().lower()
+
+    async with TaskGroup() as tg:
+        tp = tg.create_task(search_podcast(text, 'search.talkpython.fm'))
+        pb = tg.create_task(search_podcast(text, 'search.pythonbytes.fm'))
+
+    # results = await asyncio.gather(tp, pb)
+    # pprint(results)
+    # return
+
+    print("Done with search.")
+    print()
+    print("Results from Talk Python:")
+    for title in tp.result():
+        print(f'* {title}')
+    print()
+    print("Results from Python Bytes:")
+    for title in pb.result():
+        print(f'* {title}')
+
+
+if __name__ == '__main__':
+    asyncio.run(search())
diff --git a/requirements.in b/requirements.in
@@ -0,0 +1,2 @@
+httpx
+pydantic
diff --git a/requirements.txt b/requirements.txt
@@ -0,0 +1,33 @@
+#
+# This file is autogenerated by pip-compile with Python 3.11
+# by the following command:
+#
+#    pip-compile requirements.in
+#
+anyio==3.6.2
+    # via httpcore
+certifi==2022.12.7
+    # via
+    #   httpcore
+    #   httpx
+h11==0.14.0
+    # via httpcore
+httpcore==0.16.3
+    # via httpx
+httpx==0.23.1
+    # via -r requirements.in
+idna==3.4
+    # via
+    #   anyio
+    #   rfc3986
+pydantic==1.10.2
+    # via -r requirements.in
+rfc3986[idna2008]==1.5.0
+    # via httpx
+sniffio==1.3.0
+    # via
+    #   anyio
+    #   httpcore
+    #   httpx
+typing-extensions==4.4.0
+    # via pydantic
