api/kiwi_vpn_api/db_new/__init__.py

@@ -1,7 +0,0 @@
-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/capabilities.py

@@ -13,11 +13,9 @@ 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
@@ -29,8 +27,6 @@ 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

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

api/kiwi_vpn_api/main.py

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