some fun with db_new

api/kiwi_vpn_api/db_new/__init__.py

@@ -0,0 +1,7 @@
+from .capability import Capability
+from .connection import Connection
+from .device import Device, DeviceBase, DeviceCreate
+from .user import User, UserBase, UserCreate, UserRead
+
+__all__ = ["Capability", "Connection", "Device", "DeviceBase", "DeviceCreate",
+           "User", "UserBase", "UserCreate", "UserRead"]

api/kiwi_vpn_api/db_new/capability.py

@@ -13,9 +13,11 @@ class Capability(Enum):
     issue = "issue"
     renew = "renew"
 
+    def __repr__(self) -> str:
+        return self.value
+
 
 class UserCapabilityBase(SQLModel):
-    user_name: str = Field(primary_key=True, foreign_key="user.name")
     capability_name: str = Field(primary_key=True)
 
     @property
@@ -27,6 +29,8 @@ class UserCapabilityBase(SQLModel):
 
 
 class UserCapability(UserCapabilityBase, table=True):
+    user_name: str = Field(primary_key=True, foreign_key="user.name")
+
     user: "User" = Relationship(
-        back_populates="capabilities"
+        back_populates="capabilities",
     )

api/kiwi_vpn_api/db_new/device.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+from sqlalchemy.exc import IntegrityError
+from sqlmodel import Field, Relationship, SQLModel, UniqueConstraint
+
+from .connection import Connection
+
+if TYPE_CHECKING:
+    from .user import User
+
+
+class DeviceBase(SQLModel):
+    name: str
+    type: str
+    expiry: datetime | None
+
+
+class DeviceCreate(DeviceBase):
+    owner_name: str | None
+
+
+class Device(DeviceBase, table=True, ):
+    __table_args__ = (UniqueConstraint(
+        "owner_name",
+        "name",
+    ),)
+
+    id: int | None = Field(primary_key=True)
+    owner_name: str | None = Field(foreign_key="user.name")
+
+    owner: User = Relationship(
+        back_populates="devices",
+    )
+
+    @classmethod
+    def create(cls, **kwargs) -> Device | None:
+        """
+        Create a new device in the database.
+        """
+
+        try:
+            with Connection.session as db:
+                device = cls.from_orm(DeviceCreate(**kwargs))
+
+                db.add(device)
+                db.commit()
+                db.refresh(device)
+
+                return device
+
+        except IntegrityError:
+            # device already existed
+            return None
+
+    def update(self) -> None:
+        """
+        Update this device in the database.
+        """
+
+        with Connection.session as db:
+            db.add(self)
+            db.commit()
+            db.refresh(self)
+
+    def delete(self) -> bool:
+        """
+        Delete this device from the database.
+        """
+
+        with Connection.session as db:
+            db.delete(self)
+            db.commit()

api/kiwi_vpn_api/db_new/user.py

@@ -1,14 +1,15 @@
 from __future__ import annotations
 
-from typing import Any, Sequence
+from typing import Any
 
 from pydantic import root_validator
 from sqlalchemy.exc import IntegrityError
 from sqlmodel import Field, Relationship, SQLModel
 
 from ..config import Config
-from .capabilities import Capability, UserCapability
+from .capability import Capability, UserCapability
 from .connection import Connection
+from .device import Device
 
 
 class UserBase(SQLModel):
@@ -33,6 +34,10 @@ class User(UserBase, table=True):
         },
     )
 
+    devices: list[Device] = Relationship(
+        back_populates="owner",
+    )
+
     @classmethod
     def create(cls, **kwargs) -> User | None:
         """
@@ -104,34 +109,19 @@ class User(UserBase, table=True):
             db.delete(self)
             db.commit()
 
-    def extend_capabilities(self, capabilities: Sequence[Capability]) -> None:
-        """
-        Extend this user's capabilities
-        """
+    def get_capabilities(self) -> set[Capability]:
+        return set(
+            capability._
+            for capability in self.capabilities
+        )
 
-        for capability in capabilities:
-            user_capability = UserCapability(
+    def set_capabilities(self, capabilities: set[Capability]) -> None:
+        self.capabilities = [
+            UserCapability(
                 user_name=self.name,
                 capability_name=capability.value,
-            )
-
-            if user_capability not in self.capabilities:
-                self.capabilities.append(user_capability)
-
-    def remove_capabilities(self, capabilities: Sequence[Capability]) -> None:
-        """
-        Remove from this user's capabilities
-        """
-
-        for capability in capabilities:
-            try:
-                self.capabilities.remove(UserCapability(
-                    user_name=self.name,
-                    capability_name=capability.value,
-                ))
-
-            except ValueError:
-                pass
+            ) for capability in capabilities
+        ]
 
 
 class UserCreate(UserBase):

api/kiwi_vpn_api/main.py

@@ -14,8 +14,10 @@ from fastapi import FastAPI
 
 from .config import Config, Settings
 from .db import Connection
-from .db.schemata import User
-from .db_new import capabilities, connection, user
+# from .db.schemata import User
+from .db_new import Capability
+from .db_new import Connection as Connection_new
+from .db_new import Device, User
 from .routers import main_router
 
 settings = Settings.get()
@@ -47,31 +49,43 @@ async def on_startup() -> None:
         # connect to database
         Connection.connect(await current_config.db.db_engine)
 
-        # some testing
-        with Connection.use() as db:
-            print(User.from_db(db, "admin"))
-            print(User.from_db(db, "nonexistent"))
+        # # some testing
+        # with Connection.use() as db:
+        #     print(User.from_db(db, "admin"))
+        #     print(User.from_db(db, "nonexistent"))
 
-        connection.Connection.connect("sqlite:///tmp/v2.db")
+    Connection_new.connect("sqlite:///tmp/v2.db")
 
-        user.User.create(
-            name="Uwe",
-            password_clear="ulf",
-            email="uwe@feh.de",
-        )
+    User.create(
+        name="Uwe",
+        password_clear="ulf",
+        email="uwe@feh.de",
+    )
 
-        print(user.User.get("Uwe"))
-        print(user.User.authenticate("Uwe", "uwe"))
+    print(User.get(name="Uwe"))
+    print(User.authenticate("Uwe", "uwe"))
 
-        uwe = user.User.authenticate("Uwe", "ulf")
+    uwe = User.authenticate("Uwe", "ulf")
 
-        uwe.extend_capabilities([capabilities.Capability.admin])
-        uwe.update()
-        print(uwe)
+    uwe.set_capabilities([Capability.admin])
+    uwe.update()
+    print(uwe.get_capabilities())
 
-        uwe.remove_capabilities([capabilities.Capability.admin])
-        uwe.update()
-        print(uwe)
+    uwe.set_capabilities([])
+    uwe.update()
+    print(uwe.get_capabilities())
+
+    with Connection_new.session as db:
+        db.add(uwe)
+        print(uwe.devices)
+
+    ipad = Device.create(
+        owner_name="Uwe",
+        name="iPad",
+        type="tablet",
+    )
+    # ipad = Device.
+    print(ipad)
 
 
 def main() -> None: