isolate fastui

app/_internal.py

@@ -0,0 +1,41 @@
+from http import HTTPStatus  # noqa: F401
+from typing import Annotated  # noqa: F401
+
+import svcs  # noqa: F401
+from edgedb.asyncio_client import AsyncIOClient  # noqa: F401
+from fastapi import APIRouter, HTTPException, Query  # noqa: F401
+
+from .queries import (
+    get_user_by_name_with_n_events_async_edgeql as get_user_by_name_with_n_events_qry,  # noqa: F401
+)
+from .queries import get_user_with_n_events_async_edgeql as get_user_with_n_events_qry
+
+router = APIRouter()
+
+
+@router.get(
+    "/internal/users",
+    response_model=list[get_user_with_n_events_qry.GetUserWithNEventsResult]
+    | get_user_by_name_with_n_events_qry.GetUserByNameWithNEventsResult,
+    tags=["_internal"],
+)
+async def _get_users(
+    services: svcs.fastapi.DepContainer,
+    name: Annotated[str | None, Query(max_length=50)] = None,
+):
+    client = await services.aget(AsyncIOClient)
+    if name is None:
+        return await get_user_with_n_events_qry.get_user_with_n_events(client)
+    else:
+        if (
+            user
+            := await get_user_by_name_with_n_events_qry.get_user_by_name_with_n_events(
+                client, name=name
+            )
+        ):
+            return user
+
+    raise HTTPException(
+        status_code=HTTPStatus.NOT_FOUND,
+        detail={"error": f"Username '{name}' does not exist."},
+    )

app/_lifespan.py

@@ -2,12 +2,9 @@
 import svcs
 from edgedb.asyncio_client import AsyncIOClient
 from fastapi import FastAPI
-from httpx import AsyncClient
 
-from .config import settings
 
-
-async def _lifespan(app: FastAPI, registry: svcs.Registry, need_fastui: bool):
+async def _lifespan(app: FastAPI, registry: svcs.Registry):
     # EdgeDB client
     db_client = edgedb.create_async_client()
 
@@ -23,24 +20,6 @@ async def ping_db_callable(_db_client):
         ping=ping_db_callable,
     )
 
-    if need_fastui:
-        # Web client
-        host = str(settings.host).rstrip("/")  # any better ways?
-        base_url = f"{host}:{settings.port}"
-        client = AsyncClient(base_url=base_url)
-
-        async def setup_httpx_client():
-            yield client
-
-        async def ping_web_callable(_client):
-            return await _client.get("/covered-by-fastui.users.html_landing")
-
-        registry.register_factory(
-            AsyncClient,
-            setup_httpx_client,
-            ping=ping_web_callable,
-        )
-
     yield
 
     await registry.aclose()

app/events.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from http import HTTPStatus
 from typing import Annotated
 

app/health.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from http import HTTPStatus
 
 import svcs

app/lifespan.py

@@ -1,8 +1,5 @@
-from functools import partial
-
 import svcs
 
 from ._lifespan import _lifespan
-from .config import settings
 
-lifespan = svcs.fastapi.lifespan(partial(_lifespan, need_fastui=settings.need_fastui))
+lifespan = svcs.fastapi.lifespan(_lifespan)

app/main.py

@@ -1,11 +1,9 @@
-from __future__ import annotations
-
 from fastapi import FastAPI
 from starlette.middleware.cors import CORSMiddleware
 
-from app import events, health, users
-from app.config import settings
-from app.lifespan import lifespan
+from app import _internal, events, health, users
+
+from .lifespan import lifespan
 
 
 def make_app(lifespan):
@@ -23,11 +21,7 @@ def make_app(lifespan):
     app.include_router(users.router)
     app.include_router(events.router)
     app.include_router(health.router)
-
-    if settings.need_fastui:
-        from app.fastui import users as ui_users
-
-        app.include_router(ui_users.router)
+    app.include_router(_internal.router)
 
     return app
 

app/queries/get_user_by_name.edgeql

@@ -1 +1 @@
-select User {name, created_at} filter User.name=<str>$name
+select User {name, created_at} filter User.name=<str>$name;

app/queries/get_user_by_name_async_edgeql.py

@@ -39,7 +39,7 @@ async def get_user_by_name(
 ) -> GetUserByNameResult | None:
     return await executor.query_single(
         """\
-        select User {name, created_at} filter User.name=<str>$name\
+        select User {name, created_at} filter User.name=<str>$name;\
         """,
         name=name,
     )

app/queries/get_user_by_name_with_n_events.edgeql

@@ -0,0 +1,5 @@
+select User {name,
+            created_at, 
+            n_events:= count(.<host[is Event])
+        } 
+filter User.name=<str>$name;

app/queries/get_user_by_name_with_n_events_async_edgeql.py

@@ -0,0 +1,50 @@
+# AUTOGENERATED FROM 'app/queries/get_user_by_name_with_n_events.edgeql' WITH:
+#     $ edgedb-py
+
+
+from __future__ import annotations
+import dataclasses
+import datetime
+import edgedb
+import uuid
+
+
+class NoPydanticValidation:
+    @classmethod
+    def __get_pydantic_core_schema__(cls, _source_type, _handler):
+        # Pydantic 2.x
+        from pydantic_core.core_schema import any_schema
+        return any_schema()
+
+    @classmethod
+    def __get_validators__(cls):
+        # Pydantic 1.x
+        from pydantic.dataclasses import dataclass as pydantic_dataclass
+        pydantic_dataclass(cls)
+        cls.__pydantic_model__.__get_validators__ = lambda: []
+        return []
+
+
+@dataclasses.dataclass
+class GetUserByNameWithNEventsResult(NoPydanticValidation):
+    id: uuid.UUID
+    name: str
+    created_at: datetime.datetime
+    n_events: int
+
+
+async def get_user_by_name_with_n_events(
+    executor: edgedb.AsyncIOExecutor,
+    *,
+    name: str,
+) -> GetUserByNameWithNEventsResult | None:
+    return await executor.query_single(
+        """\
+        select User {name,
+                    created_at, 
+                    n_events:= count(.<host[is Event])
+                } 
+        filter User.name=<str>$name;\
+        """,
+        name=name,
+    )

app/queries/get_user_with_n_events.edgeql

@@ -0,0 +1,4 @@
+select User {name,
+            created_at, 
+            n_events:= count(.<host[is Event])
+        } ;

app/queries/get_user_with_n_events_async_edgeql.py

@@ -0,0 +1,46 @@
+# AUTOGENERATED FROM 'app/queries/get_user_with_n_events.edgeql' WITH:
+#     $ edgedb-py
+
+
+from __future__ import annotations
+import dataclasses
+import datetime
+import edgedb
+import uuid
+
+
+class NoPydanticValidation:
+    @classmethod
+    def __get_pydantic_core_schema__(cls, _source_type, _handler):
+        # Pydantic 2.x
+        from pydantic_core.core_schema import any_schema
+        return any_schema()
+
+    @classmethod
+    def __get_validators__(cls):
+        # Pydantic 1.x
+        from pydantic.dataclasses import dataclass as pydantic_dataclass
+        pydantic_dataclass(cls)
+        cls.__pydantic_model__.__get_validators__ = lambda: []
+        return []
+
+
+@dataclasses.dataclass
+class GetUserWithNEventsResult(NoPydanticValidation):
+    id: uuid.UUID
+    name: str
+    created_at: datetime.datetime
+    n_events: int
+
+
+async def get_user_with_n_events(
+    executor: edgedb.AsyncIOExecutor,
+) -> list[GetUserWithNEventsResult]:
+    return await executor.query(
+        """\
+        select User {name,
+                    created_at, 
+                    n_events:= count(.<host[is Event])
+                } ;\
+        """,
+    )

app/users.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from http import HTTPStatus
 from typing import Annotated
 

fastui_app/_lifespan.py

@@ -0,0 +1,24 @@
+import svcs
+from fastapi import FastAPI
+from httpx import AsyncClient
+
+from fastui_app.config import settings
+
+
+async def _lifespan(app: FastAPI, registry: svcs.Registry):
+    # Web client(connect to backend)
+    host = str(settings.backendhost).rstrip("/")  # any better ways?
+    base_url = f"{host}:{settings.backendport}"
+    client = AsyncClient(base_url=base_url)
+
+    async def setup_httpx_client():
+        yield client
+
+    registry.register_factory(
+        AsyncClient,
+        setup_httpx_client,
+    )
+
+    yield
+
+    await registry.aclose()

fastui_app/config.py

@@ -0,0 +1,18 @@
+from pydantic import HttpUrl
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    frontendhost: HttpUrl = "http://localhost"
+    frontendport: int = 7000
+
+    backendhost: HttpUrl = "http://localhost"
+    backendport: int = 8000
+
+    model_config = SettingsConfigDict(
+        env_file="frontend.env", env_file_encoding="utf-8"
+    )
+
+
+settings = Settings()
+print(f"{settings=} from frontend")

fastui_app/lifespan.py

@@ -0,0 +1,5 @@
+import svcs
+
+from ._lifespan import _lifespan
+
+lifespan = svcs.fastapi.lifespan(_lifespan)

fastui_app/main.py

@@ -0,0 +1,26 @@
+from fastapi import FastAPI
+from starlette.middleware.cors import CORSMiddleware
+
+from fastui_app import users
+
+from .lifespan import lifespan
+
+
+def make_app(lifespan):
+    app = FastAPI(lifespan=lifespan)
+
+    # Set all CORS enabled origins
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    app.include_router(users.router)
+
+    return app
+
+
+app = make_app(lifespan)