Merge pull request #3 from sinisaos/use_pydantic

use Pydantic to parse yaml

README.md

@@ -39,15 +39,15 @@ Example of `config.yaml`:
 ```yaml
 tables:
   # An example of additional Piccolo Admin configuration
-  Actor:
+  - table_name: actor
     visible_columns:
       - first_name
     visible_filters:
       - actor_id
       - first_name
     menu_group: Movies
     link_column: first_name
-  Address:
+  - table_name: address
     visible_columns:
       - address_id
       - address
@@ -57,23 +57,23 @@ tables:
       - address
       - city_id
     menu_group: Location
-    rich_text_columns: address
-  City:
+    rich_text_columns: 
+      - address
+  - table_name: city
     visible_columns:
       - city_id
       - city
     visible_filters:
       - city_id
       - city
     menu_group: Location
-  Country:
+  - table_name: country
     visible_columns:
       - country_id
       - country
     visible_filters:
       - country_id
       - country
-    menu_group: Location
 
 sidebar_links:
   Piccolo Admin: https://piccolo-admin.readthedocs.io/en/latest/index.html

app/main.py

@@ -1,42 +1,16 @@
 import asyncio
 import os
 
-import yaml
 from hypercorn import Config
 from hypercorn.asyncio import serve
-from piccolo.apps.user.tables import BaseUser
 from piccolo.engine import PostgresEngine
-from piccolo.engine.sqlite import SQLiteEngine
 from piccolo.table import create_db_tables
 from piccolo.table_reflection import TableStorage
 from piccolo_admin.endpoints import TableConfig, create_admin
 from piccolo_api.encryption.providers import XChaCha20Provider
 from piccolo_api.mfa.authenticator.provider import AuthenticatorProvider
-from piccolo_api.mfa.authenticator.tables import (
-    AuthenticatorSecret as AuthenticatorSecret_,
-)
-from piccolo_api.session_auth.tables import SessionsBase
-
-with open("config.yaml") as stream:
-    try:
-        admin_config = yaml.safe_load(stream)
-        BASE_CONFIG = admin_config.get("tables")
-    except yaml.YAMLError as exc:
-        raise exc
-
-DB = SQLiteEngine()
-
-
-class Sessions(SessionsBase, db=DB):
-    pass
-
-
-class User(BaseUser, tablename="piccolo_user", db=DB):
-    pass
-
-
-class AuthenticatorSecret(AuthenticatorSecret_, db=DB):
-    pass
+from tables import AuthenticatorSecret, Sessions, User
+from utils import additional_config
 
 
 async def main():
@@ -72,79 +46,55 @@ async def main():
     await storage.reflect(schema_name="public")
 
     # additional configuration of admin tables
-    if BASE_CONFIG is not None:
-        tables_to_show = [table.lower() for table in BASE_CONFIG]
-        found_tables = [
-            table
-            for table in storage.tables.values()
-            if table._meta.tablename in tables_to_show
-        ]
+    if additional_config.tables is not None:
         admin_tables = []
-        for table in found_tables:
-            capitalize_table_name = table._meta.tablename.capitalize()
-            # visible columns
-            try:
-                visible_columns = [
-                    column
-                    for column in table._meta.columns
-                    if column._meta.name
-                    in BASE_CONFIG[capitalize_table_name].get(
-                        "visible_columns", None
-                    )
-                ]
-            except TypeError:
-                visible_columns = None
-            # visible filters
-            try:
-                visible_filters = [
-                    column
-                    for column in table._meta.columns
-                    if column._meta.name
-                    in BASE_CONFIG[capitalize_table_name].get(
-                        "visible_filters", None
+        for config in additional_config.tables:
+            for table in storage.tables.values():
+                if table._meta.tablename == config.table_name:
+                    # visible columns
+                    if config.visible_columns:
+                        visible_columns = [
+                            table._meta.get_column_by_name(column_name)
+                            for column_name in config.visible_columns
+                        ]
+                    else:
+                        visible_columns = table._meta.columns
+                    # visible filters
+                    if config.visible_filters:
+                        visible_filters = [
+                            table._meta.get_column_by_name(column_name)
+                            for column_name in config.visible_filters
+                        ]
+                    else:
+                        visible_filters = table._meta.columns
+                    # rich text columns
+                    if config.rich_text_columns:
+                        rich_text_columns = [
+                            table._meta.get_column_by_name(column_name)
+                            for column_name in config.rich_text_columns
+                        ]
+                    else:
+                        rich_text_columns = None
+                    # link column
+                    if config.link_column:
+                        link_column = table._meta.get_column_by_name(
+                            config.link_column
+                        )
+                    else:
+                        link_column = None
+                    # menu_group
+                    menu_group = config.menu_group
+
+                    admin_tables.append(
+                        TableConfig(
+                            table_class=table,
+                            visible_columns=visible_columns,
+                            visible_filters=visible_filters,
+                            rich_text_columns=rich_text_columns,
+                            link_column=link_column,
+                            menu_group=menu_group,
+                        )
                     )
-                ]
-            except TypeError:
-                visible_filters = None
-            # rich text columns
-            try:
-                rich_text_columns = [
-                    column
-                    for column in table._meta.columns
-                    if column._meta.name
-                    in BASE_CONFIG[capitalize_table_name].get(
-                        "rich_text_columns", None
-                    )
-                ]
-            except TypeError:
-                rich_text_columns = None
-            # link column
-            try:
-                link_column = [
-                    column
-                    for column in table._meta.columns
-                    if column._meta.name
-                    == BASE_CONFIG[capitalize_table_name].get(
-                        "link_column", None
-                    )
-                ][0]
-            except IndexError:
-                link_column = None
-            # menu_group
-            menu_group = BASE_CONFIG[capitalize_table_name].get(
-                "menu_group", None
-            )
-
-            admin_tables.append(
-                TableConfig(
-                    table_class=table,
-                    visible_columns=visible_columns,
-                    visible_filters=visible_filters,
-                    rich_text_columns=rich_text_columns,
-                    link_column=link_column,
-                    menu_group=menu_group,
-                )
-            )
     else:
         admin_tables = storage.tables.values()
 
@@ -170,7 +120,7 @@ async def main():
                 secret_table=AuthenticatorSecret,
             ),
         ],
-        sidebar_links=admin_config.get("sidebar_links", None),
+        sidebar_links=additional_config.sidebar_links or {},
     )
 
     # Server

app/models.py

@@ -0,0 +1,17 @@
+import typing as t
+
+from pydantic import BaseModel
+
+
+class AdminTable(BaseModel):
+    table_name: t.Optional[str] = None
+    visible_columns: t.Optional[list[str]] = None
+    visible_filters: t.Optional[list[str]] = None
+    rich_text_columns: t.Optional[list[str]] = None
+    link_column: t.Optional[str] = None
+    menu_group: t.Optional[str] = None
+
+
+class AdditionalConfig(BaseModel):
+    tables: t.Optional[list[AdminTable]] = None
+    sidebar_links: t.Optional[dict[str, str]] = None

app/tables.py

@@ -0,0 +1,20 @@
+from piccolo.apps.user.tables import BaseUser
+from piccolo.engine.sqlite import SQLiteEngine
+from piccolo_api.mfa.authenticator.tables import (
+    AuthenticatorSecret as AuthenticatorSecret_,
+)
+from piccolo_api.session_auth.tables import SessionsBase
+
+DB = SQLiteEngine()
+
+
+class Sessions(SessionsBase, db=DB):
+    pass
+
+
+class User(BaseUser, tablename="piccolo_user", db=DB):
+    pass
+
+
+class AuthenticatorSecret(AuthenticatorSecret_, db=DB):
+    pass

app/utils.py

@@ -0,0 +1,11 @@
+import yaml
+from models import AdditionalConfig
+
+
+def load_yaml() -> AdditionalConfig:
+    with open("config.yaml") as file:
+        admin_config = yaml.safe_load(file)
+        return AdditionalConfig(**admin_config)
+
+
+additional_config = load_yaml()