User.create() methods

api/kiwi_vpn_api/db/__init__.py

@@ -3,7 +3,7 @@ Package `db`: ORM and schemas for database content.
 """
 
 from .connection import Connection
-from .device import Device, DeviceBase, DeviceCreate
+from .device import Device, DeviceBase, DeviceCreate, DeviceRead
 from .user import User, UserBase, UserCreate, UserRead
 from .user_capability import UserCapabilityType
 
@@ -12,6 +12,7 @@ __all__ = [
     "Device",
     "DeviceBase",
     "DeviceCreate",
+    "DeviceRead",
     "User",
     "UserBase",
     "UserCreate",

api/kiwi_vpn_api/db/device.py

@@ -31,8 +31,6 @@ class DeviceCreate(DeviceBase):
     Representation of a newly created device
     """
 
-    owner_name: str | None
-
 
 class DeviceRead(DeviceBase):
     """
@@ -62,25 +60,47 @@ class Device(DeviceBase, table=True):
     )
 
     @classmethod
-    def create(cls, **kwargs) -> Device | None:
+    def create(
+        cls,
+        *,
+        owner: User,
+        device: DeviceCreate,
+    ) -> Device | None:
         """
         Create a new device in the database.
         """
 
         try:
             with Connection.session as db:
-                device = cls.from_orm(DeviceCreate(**kwargs))
+                new_device = cls.from_orm(device)
+                new_device.owner = owner
 
-                db.add(device)
+                db.add(new_device)
                 db.commit()
-                db.refresh(device)
+                db.refresh(new_device)
 
-                return device
+                return new_device
 
         except IntegrityError:
             # device already existed
             return None
 
+    @classmethod
+    def create_kwargs(
+        cls,
+        *,
+        owner: User,
+        **kwargs
+    ) -> Device | None:
+        """
+        Create a new device in the database. Keywords version.
+        """
+
+        return cls.create(
+            owner=owner,
+            device=DeviceCreate(**kwargs),
+        )
+
     def update(self) -> None:
         """
         Update this device in the database.

api/kiwi_vpn_api/db/user.py

@@ -90,25 +90,42 @@ class User(UserBase, table=True):
     )
 
     @classmethod
-    def create(cls, **kwargs) -> User | None:
+    def create(
+        cls,
+        *,
+        user: UserCreate,
+    ) -> User | None:
         """
         Create a new user in the database.
         """
 
         try:
             with Connection.session as db:
-                user = cls.from_orm(UserCreate(**kwargs))
+                new_user = cls.from_orm(user)
 
-                db.add(user)
+                db.add(new_user)
                 db.commit()
-                db.refresh(user)
+                db.refresh(new_user)
 
-                return user
+                return new_user
 
         except IntegrityError:
             # user already existed
             return None
 
+    @classmethod
+    def create_kwargs(
+        cls,
+        **kwargs
+    ) -> User | None:
+        """
+        Create a new user in the database. Keywords version.
+        """
+
+        return cls.create(
+            user=UserCreate(**kwargs),
+        )
+
     @classmethod
     def get(cls, name: str) -> User | None:
         """

api/kiwi_vpn_api/routers/__init__.py

@@ -6,11 +6,12 @@ This file: Main API router definition.
 
 from fastapi import APIRouter
 
-from . import admin, user
+from . import admin, device, user
 
 main_router = APIRouter()
 
 main_router.include_router(admin.router)
+main_router.include_router(device.router)
 main_router.include_router(user.router)
 
 __all__ = [

api/kiwi_vpn_api/routers/_common.py

@@ -40,7 +40,7 @@ class Responses:
         "description": "Must be admin",
         "content": None,
     }
-    NEEDS_ADMIN_OR_SELF = {
+    NEEDS_REQUESTED_USER = {
         "description": "Must be the requested user",
         "content": None,
     }
@@ -99,20 +99,29 @@ async def get_current_user_if_admin(
     return current_user
 
 
-async def get_current_user_if_admin_or_self(
+async def get_user_by_name(
     user_name: str,
     current_user: User = Depends(get_current_user_if_exists),
 ) -> User:
     """
-    Get the currently logged-in user.
+    Get a user by name.
 
-    Fails a) if the currently logged-in user is not the requested user,
+    Works if a) the currently logged-in user is an admin,
-    and b) if it is not an admin.
+    or b) if it is the requested user.
     """
 
-    # fail if not requested by an admin or self
+    # check if current user is admin
-    if not (current_user.can(UserCapabilityType.admin)
+    if current_user.can(UserCapabilityType.admin):
-            or current_user.name == user_name):
+        # fail if requested user doesn't exist
+        if (user := User.get(user_name)) is None:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
+
+    # check if current user is requested user
+    elif current_user.name == user_name:
+        pass
+
+    # current user is neither admin nor the requested user
+    else:
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
 
-    return current_user
+    return user

api/kiwi_vpn_api/routers/admin.py

@@ -63,7 +63,7 @@ async def create_initial_admin(
             raise HTTPException(status_code=status.HTTP_409_CONFLICT)
 
     # create an administrative user
-    new_user = User.create(**admin_user.dict())
+    new_user = User.create(admin_user)
     new_user.set_capabilities([UserCapabilityType.admin])
     new_user.update()
 

api/kiwi_vpn_api/routers/device.py

@@ -0,0 +1,44 @@
+"""
+/device endpoints.
+"""
+
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from ..db import Device, DeviceCreate, DeviceRead, User
+from ._common import Responses, get_user_by_name
+
+router = APIRouter(prefix="/device", tags=["device"])
+
+
+@router.post(
+    "",
+    responses={
+        status.HTTP_200_OK: Responses.OK,
+        status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
+        status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
+        status.HTTP_403_FORBIDDEN: Responses.NEEDS_REQUESTED_USER,
+        status.HTTP_404_NOT_FOUND: Responses.ENTRY_DOESNT_EXIST,
+        status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
+    },
+    response_model=DeviceRead,
+)
+async def add_device(
+    device: DeviceCreate,
+    user: User = Depends(get_user_by_name),
+) -> Device:
+    """
+    POST ./: Create a new device in the database.
+    """
+
+    # create the new device
+    new_device = Device.create(
+        owner=user,
+        device=device,
+    )
+
+    # fail if creation was unsuccessful
+    if new_device is None:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT)
+
+    # return the created device on success
+    return new_device

api/kiwi_vpn_api/routers/user.py

@@ -76,24 +76,26 @@ async def get_current_user(
         status.HTTP_403_FORBIDDEN: Responses.NEEDS_ADMIN,
         status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
     },
-    response_model=User,
+    response_model=UserRead,
 )
 async def add_user(
     user: UserCreate,
     _: User = Depends(get_current_user_if_admin),
-):
+) -> User:
     """
     POST ./: Create a new user in the database.
     """
 
     # actually create the new user
-    new_user = User.create(**user.dict())
+    new_user = User.create(user=user)
-    new_user.set_capabilities([UserCapabilityType.login])
 
     # fail if creation was unsuccessful
     if new_user is None:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT)
 
+    new_user.set_capabilities([UserCapabilityType.login])
+    new_user.update()
+
     # return the created user on success
     return new_user