make chart data dynamic

Author: dantownsend

piccolo_admin/endpoints.py

@@ -3,6 +3,7 @@
 """
 from __future__ import annotations
 
+import decimal
 import inspect
 import itertools
 import json
@@ -349,7 +350,10 @@ class FormConfigResponseModel(BaseModel):
     description: t.Optional[str] = None
 
 
-@dataclass
+Number = t.Union[int, float, decimal.Decimal]
+ChartData = t.Sequence[t.Tuple[str, Number]]
+
+
 class ChartConfig:
     """
     Used to specify charts, which are passed into ``create_admin``.
@@ -361,46 +365,58 @@ class ChartConfig:
     :param chart_type:
         Available chart types. There are five types: ``Pie``, ``Line``,
         ``Column``, ``Bar`` and ``Area``.
-    :param data:
-        The data to be passed to the admin ui. The data format must be
-        a ``list of lists`` (eg. ``[["Male", 7], ["Female", 3]]``).
+    :param data_source:
+        A function (async or sync) which returns the data to be displayed in
+        the chart. It must returns a sequence of tuples. The first element is
+        the label, and the second is the value::
+
+            >>> [("Male", 7), ("Female", 3)]
 
     Here's a full example:
 
     .. code-block:: python
 
-        async def director_movie_count():
+        async def get_director_movie_count():
             movies = await Movie.select(
                 Movie.director.name.as_alias("director"),
                 Count(Movie.id)
             ).group_by(
                 Movie.director
             )
             # Flatten the response so it's a list of lists
-            # like [['George Lucas', 3], ...]
-            return [[i['director'], i['count']] for i in movies]
+            # like [('George Lucas', 3), ...]
+            return [(i['director'], i['count']) for i in movies]
 
         director_chart = ChartConfig(
             title='Movie count',
-            chart_type="Pie", # or Bar or Line etc.
-            data=director_movie_count,
+            chart_type="Pie",  # or Bar or Line etc.
+            data_source=get_director_movie_count
+        )
 
         create_admin(charts=[director_chart])
 
     """
 
-    def __init__(self, title: str, chart_type: str, data: t.List[t.Any]):
+    def __init__(
+        self,
+        title: str,
+        data_source: t.Callable[[], t.Awaitable[ChartData]],
+        chart_type: t.Literal["Pie", "Line", "Column", "Bar", "Area"] = "Bar",
+    ):
         self.title = title
         self.chart_slug = self.title.replace(" ", "-").lower()
         self.chart_type = chart_type
-        self.data = data
+        self.data_source = data_source
 
 
-class ChartConfigResponseModel(BaseModel):
+class ChartResponseModel(BaseModel):
     title: str
     chart_slug: str
     chart_type: str
-    data: t.List[t.Any]
+
+
+class ChartDataResponseModel(ChartResponseModel):
+    data: ChartData
 
 
 def handle_auth_exception(request: Request, exc: Exception):
@@ -440,7 +456,7 @@ class AdminRouter(FastAPI):
     def __init__(
         self,
         *tables: t.Union[t.Type[Table], TableConfig],
-        forms: t.List[FormConfig] = [],
+        forms: t.Optional[t.List[FormConfig]] = None,
         auth_table: t.Type[BaseUser] = BaseUser,
         session_table: t.Type[SessionsBase] = SessionsBase,
         session_expiry: timedelta = timedelta(hours=1),
@@ -452,11 +468,11 @@ def __init__(
         production: bool = False,
         site_name: str = "Piccolo Admin",
         default_language_code: str = "auto",
-        translations: t.List[Translation] = None,
+        translations: t.Optional[t.List[Translation]] = None,
         allowed_hosts: t.Sequence[str] = [],
         debug: bool = False,
-        charts: t.List[ChartConfig] = [],
-        sidebar_links: t.Dict[str, str] = {},
+        charts: t.Optional[t.List[ChartConfig]] = None,
+        sidebar_links: t.Optional[t.Dict[str, str]] = None,
     ) -> None:
         super().__init__(
             title=site_name,
@@ -538,13 +554,13 @@ def __init__(
 
         self.auth_table = auth_table
         self.site_name = site_name
-        self.forms = forms
+        self.forms = forms or []
         self.read_only = read_only
-        self.charts = charts
+        self.charts = charts or []
         self.chart_config_map = {
             chart.chart_slug: chart for chart in self.charts
         }
-        self.sidebar_links = sidebar_links
+        self.sidebar_links = sidebar_links or {}
         self.form_config_map = {form.slug: form for form in self.forms}
 
         with open(os.path.join(ASSET_PATH, "index.html")) as f:
@@ -657,15 +673,15 @@ def __init__(
             endpoint=self.get_charts,  # type: ignore
             methods=["GET"],
             tags=["Charts"],
-            response_model=t.List[ChartConfigResponseModel],
+            response_model=t.List[ChartResponseModel],
         )
 
         private_app.add_api_route(
             path="/charts/{chart_slug:str}/",
             endpoint=self.get_single_chart,  # type: ignore
             methods=["GET"],
             tags=["Charts"],
-            response_model=ChartConfigResponseModel,
+            response_model=ChartDataResponseModel,
         )
 
         private_app.add_api_route(
@@ -923,33 +939,35 @@ def get_user(self, request: Request) -> UserResponseModel:
     ###########################################################################
     # Custom charts
 
-    def get_charts(self) -> t.List[ChartConfigResponseModel]:
+    def get_charts(self) -> t.List[ChartResponseModel]:
         """
         Returns  all charts registered with the admin.
         """
         return [
-            ChartConfigResponseModel(
+            ChartResponseModel(
                 title=chart.title,
                 chart_slug=chart.chart_slug,
                 chart_type=chart.chart_type,
-                data=chart.data,
             )
             for chart in self.charts
         ]
 
-    def get_single_chart(self, chart_slug: str) -> ChartConfigResponseModel:
+    async def get_single_chart(
+        self, chart_slug: str
+    ) -> ChartDataResponseModel:
         """
-        Returns single chart.
+        Returns a single chart.
         """
         chart = self.chart_config_map.get(chart_slug, None)
         if chart is None:
             raise HTTPException(status_code=404, detail="No such chart found")
         else:
-            return ChartConfigResponseModel(
+            data = await chart.data_source()
+            return ChartDataResponseModel(
                 title=chart.title,
                 chart_slug=chart.chart_slug,
                 chart_type=chart.chart_type,
-                data=chart.data,
+                data=data,
             )
 
     ###########################################################################
@@ -1156,7 +1174,7 @@ def create_admin(
     production: bool = False,
     site_name: str = "Piccolo Admin",
     default_language_code: str = "auto",
-    translations: t.List[Translation] = None,
+    translations: t.Optional[t.List[Translation]] = None,
     auto_include_related: bool = True,
     allowed_hosts: t.Sequence[str] = [],
     debug: bool = False,

piccolo_admin/example.py

@@ -40,6 +40,7 @@
 from piccolo.columns.readable import Readable
 from piccolo.engine.postgres import PostgresEngine
 from piccolo.engine.sqlite import SQLiteEngine
+from piccolo.query.methods.select import Count as CountAgg
 from piccolo.table import Table, create_db_tables_sync, drop_db_tables_sync
 from piccolo_api.media.local import LocalMediaStorage
 from piccolo_api.media.s3 import S3MediaStorage
@@ -418,6 +419,29 @@ def booking_endpoint(request: Request, data: BookingModel) -> str:
     menu_group="Testing",
 )
 
+
+async def get_director_movie_count():
+    movies = (
+        await Movie.select(
+            Movie.director.name.as_alias("director"), CountAgg(Movie.id)
+        )
+        .group_by(Movie.director)
+        .order_by(Movie.director.name)
+    )
+
+    return [(i["director"], i["count"]) for i in movies]
+
+
+async def get_movie_genre_count():
+    movies = (
+        await Movie.select(Movie.genre, CountAgg(Movie.id))
+        .group_by(Movie.genre)
+        .order_by(Movie.genre)
+    )
+
+    return [(i["genre"], i["count"]) for i in movies]
+
+
 APP = create_admin(
     [
         movie_config,
@@ -445,18 +469,14 @@ def booking_endpoint(request: Request, data: BookingModel) -> str:
     session_table=Sessions,
     charts=[
         ChartConfig(
-            title="Movie count",
+            title="Movies per director",
             chart_type="Pie",
-            data=[
-                ["George Lucas", 4],
-                ["Peter Jackson", 6],
-                ["Ron Howard", 1],
-            ],
+            data_source=get_director_movie_count,
         ),
         ChartConfig(
-            title="Director gender",
+            title="Movies per genre",
             chart_type="Column",
-            data=[["Male", 7], ["Female", 3]],
+            data_source=get_movie_genre_count,
         ),
     ],
     sidebar_links={

requirements/dev-requirements.txt

@@ -1,7 +1,7 @@
-black==21.12b0
-isort==5.10.1
-twine==4.0.0
-mypy==0.942
+black==23.1a1
+isort==5.12.0
+twine==4.0.2
+mypy==1.4.1
 pip-upgrader==1.4.15
-wheel==0.37.1
-python-dotenv==0.20.0
+wheel==0.40.0
+python-dotenv==1.0.0