FastAPI + PonyORM example (#59)

Author: solfrii

examples/fastapi_ponyorm/Makefile

@@ -0,0 +1,9 @@
+all: install run
+
+.PHONY: fastapi
+run:
+	poetry run fastapi dev --reload --port=8090 example.py
+
+.PHONY: install
+install:
+	poetry install

examples/fastapi_ponyorm/README.md

@@ -0,0 +1,11 @@
+## FastAPI + Pony ORM Example
+
+- Uses in-memory SQLite 3 instance
+- Creates "admin/admin" superuser
+
+```bash
+make install  # Creates virtualenv with Poetry
+make run      # Runs fastapi dev
+```
+
+So open `http://127.0.0.1:8090/admin/` and have a fun!

examples/fastapi_ponyorm/__init__.py

@@ -0,0 +1,9 @@
+import os
+import sys
+from pathlib import Path
+
+os.environ["ADMIN_USER_MODEL"] = "User"
+os.environ["ADMIN_USER_MODEL_USERNAME_FIELD"] = "username"
+os.environ["ADMIN_SECRET_KEY"] = "secret"
+
+sys.path.append(str(Path(__file__).resolve().parent))

examples/fastapi_ponyorm/example.py

@@ -0,0 +1,109 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from models import db, BaseEvent, Event, Tournament, User
+from pony.orm import db_session
+
+from fastadmin import PonyORMInlineModelAdmin, PonyORMModelAdmin, action, display
+from fastadmin import fastapi_app as admin_app
+from fastadmin import register
+
+
+@register(User)
+class UserModelAdmin(PonyORMModelAdmin):
+    exclude = ("password",)
+    list_display = ("id", "username", "is_superuser")
+    list_display_links = ("id", "username")
+    list_filter = ("id", "username", "is_superuser")
+    search_fields = ("username",)
+
+    async def authenticate(self, username, password):
+        with db_session:
+            obj = next((f for f in User.select(username=username, password=password, is_superuser=True)), None)  # fmt: skip
+            if not obj:
+                return None
+            return obj.id
+
+    async def change_password(self, user_id, password):
+        user = await self.model_cls.filter(id=user_id).first()
+        if not user:
+            return
+        # direct saving password is only for tests - use hash
+        user.password = password
+        await user.save()
+
+
+class EventInlineModelAdmin(PonyORMInlineModelAdmin):
+    model = Event
+
+
+@register(Tournament)
+class TournamentModelAdmin(PonyORMModelAdmin):
+    list_display = ("id", "name")
+    inlines = (EventInlineModelAdmin,)
+
+
+@register(BaseEvent)
+class BaseEventModelAdmin(PonyORMModelAdmin):
+    pass
+
+
+@register(Event)
+class EventModelAdmin(PonyORMModelAdmin):
+    actions = ("make_is_active", "make_is_not_active")
+    list_display = (
+        "id",
+        "name_with_price",
+        "rating",
+        "event_type",
+        "is_active",
+        "started",
+    )
+
+    @action(description="Make user active")
+    async def make_is_active(self, ids):
+        await self.model_cls.filter(id__in=ids).update(is_active=True)
+
+    @action
+    async def make_is_not_active(self, ids):
+        await self.model_cls.filter(id__in=ids).update(is_active=False)
+
+    @display
+    async def started(self, obj):
+        return bool(obj.start_time)
+
+    @display()
+    async def name_with_price(self, obj):
+        return f"{obj.name} - {obj.price}"
+
+
+def init_db():
+    db.bind(provider="sqlite", filename=":sharedmemory:")
+    db.generate_mapping(create_tables=True)
+    pass
+
+
+@db_session
+def create_superuser():
+    User(username="admin", password="admin", is_superuser=True)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    init_db()
+    create_superuser()
+    yield
+
+
+app = FastAPI(lifespan=lifespan)
+
+app.mount("/admin", admin_app)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)

examples/fastapi_ponyorm/models.py

@@ -0,0 +1,87 @@
+from datetime import date, datetime, time
+from decimal import Decimal
+from enum import Enum
+
+from pony.orm import *
+
+
+db = Database()
+
+
+class EventTypeEnum(str, Enum):
+    PRIVATE = "PRIVATE"
+    PUBLIC = "PUBLIC"
+
+
+class BaseModel:
+    # id = PrimaryKey(int, auto=True)
+    created_at = Required(datetime, default=datetime.utcnow)
+    updated_at = Required(datetime, default=datetime.utcnow)
+
+    def before_update(self):
+        self.updated_at = datetime.utcnow()
+
+    @classmethod
+    def get_model_name(cls):
+        return f"ponyorm.{cls.__name__}"
+
+
+class User(db.Entity, BaseModel):
+    _table_ = "user"
+
+    username = Required(str, max_len=255)
+    password = Required(str, max_len=255)
+    is_superuser = Required(bool, default=False)
+
+    events = Set("Event", table="event_participants", column="event_id")
+
+    def __str__(self):
+        return self.username
+
+
+class Tournament(db.Entity, BaseModel):
+    _table_ = "tournament"
+
+    name = Required(str, max_len=255)
+
+    events = Set("Event")
+
+    def __str__(self):
+        return self.name
+
+
+class BaseEvent(db.Entity, BaseModel):
+    _table_ = "base_event"
+
+    id = PrimaryKey(int, auto=True)
+    name = Required(str, max_len=255)
+
+    event = Optional("Event")
+
+    def __str__(self):
+        return self.name
+
+
+class Event(db.Entity, BaseModel):
+    _table_ = "event"
+
+    base = Optional(BaseEvent, column="base_id")
+    name = Required(str)
+
+    tournament = Required(Tournament, column="tournament_id")
+    participants = Set(User, table="event_participants", column="user_id")
+
+    rating = Required(int, default=0)
+    description = Optional(LongStr)
+    event_type = Required(EventTypeEnum, default=EventTypeEnum.PUBLIC)
+    is_active = Required(bool, default=True)
+    start_time = Optional(time)
+    date = Optional(date)
+    latitude = Optional(float)
+    longitude = Optional(float)
+    price = Optional(Decimal)
+
+    json = Optional(Json)
+
+    def __str__(self):
+        return self.name