app/controllers/aaManager.py

@@ -15,11 +15,18 @@ aa_router = APIRouter(prefix="/api/v1/aa", tags=["AccessAuth"])
 @aa_router.post("/", response_model=AccessAuthorizationResponse)
 def add_accessauth(*, db: Session = Depends(get_session), aa: AccessAuthorizationCreate, admin: bool = Depends(auth_is_admin)):
     logger.info(f"Creating accessauth with data: {aa}")
-    timetables = [Timetable.model_validate(t) for t in aa.timetables]
+    if aa.timetables is not []:
+        timetables = [Timetable.model_validate(t) for t in aa.timetables]
+    else: timetables = []
+    if aa.oneshot is not None:
+        oneshot = OneShotAccess.model_validate(aa.oneshot)
+    else: oneshot = None
     db_aa = AccessAuthorizationDB(
         name=aa.name,
+        type=aa.type,
         is_active=aa.is_active,
-        timetables=timetables
+        timetables=timetables,
+        oneshot=oneshot
     )
     return add_and_refresh(db, db_aa)
 
@@ -68,12 +75,16 @@ def change_accessauth(*, db: Session = Depends(get_session), aa_id: int, aa: Acc
     db_aa = db.get(AccessAuthorizationDB, aa_id)
     if db_aa is None:
         raise HTTPException(status_code=404, detail="AccessAuthorization not found")
-    aa_data = aa.model_dump(exclude_unset=True)
+    aa_data = aa.model_dump(exclude_unset=True, exclude_none=True)
     if "timetables" in aa_data and aa_data["timetables"] is not None:
         db_aa.timetables.clear()
         timetables = [Timetable.model_validate(t) for t in aa_data["timetables"]]
         db_aa.timetables = timetables
         aa_data.pop("timetables")
+    if "oneshot" in aa_data and aa_data["oneshot"] is not None:
+        oneshot = OneShotAccess.model_validate(aa_data["oneshot"])
+        db_aa.oneshot = oneshot
+        aa_data.pop("oneshot")
     db_aa.sqlmodel_update(aa_data)
     return add_and_refresh(db, db_aa)
 

app/model/models.py

@@ -1,6 +1,7 @@
 from sqlmodel import Field, Relationship, Session, SQLModel
-from typing import List
+from typing import List, Literal, Union
-from datetime import time
+from datetime import datetime, time
+from pydantic import model_validator
 
 class Base(SQLModel):
     pass
@@ -60,25 +61,57 @@ class GroupResponse(GroupBase):
 #### AccessAuthorization
 class AccessAuthorizationBase(Base):
     name: str = Field(index=True)
+    type: Literal["timetable", "oneshot", "somefuturespec"]
     is_active: bool
 
 class AccessAuthorizationDB(AccessAuthorizationBase, table=True):
     id: int | None = Field(default=None, primary_key=True)
+    type: str
     groups: List["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink)
     timetables: List["Timetable"] = Relationship(back_populates="accessauth", cascade_delete=True)
+    oneshot: "OneShotAccess" = Relationship(back_populates="accessauth", cascade_delete=True)
+
+class OneShotAccessBase(Base):
+    uses: int = 1
+    ends_at: datetime
+
+class OneShotAccess(OneShotAccessBase, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    accessauth_id: int = Field(default=None, foreign_key="accessauthorizationdb.id")
+    accessauth: AccessAuthorizationDB = Relationship(back_populates="oneshot")
 
 class AccessAuthorizationCreate(AccessAuthorizationBase):
-    timetables: List["TimetableCreate"]
+    timetables: List["TimetableCreate"] = []
+    oneshot: OneShotAccessBase | None = None
+    
+    @model_validator(mode="after")
+    def check_type(self):
+        if self.type == "timetable":
+            if not self.timetables:
+                raise ValueError("timetable auths require at least one timetable object")
+            if self.oneshot is not None:
+                raise ValueError("timetable auths are not allowed oneshot objects")
+        elif self.type == "oneshot":
+            if not self.oneshot:
+                raise ValueError("oneshot auths require a oneshot object")
+            if self.timetables:
+                raise ValueError("oneshot auths are not allowed timetable objects")
+        elif self.type == "somefuturespec":
+            raise ValueError("somefuturespec is not jet implemented. Please do not use")
+        return self
 
 class AccessAuthorizationResponse(AccessAuthorizationBase):
     id: int
-    timetables: List["Timetable"]
+    timetables: List["Timetable"] = []
+    oneshot: OneShotAccessBase | None = None
     groups: List["GroupDB"]
 
 class AccessAuthorizationUpdate(Base):
     name: str | None = None
+    type: Literal["timetable", "oneshot", "somefuturespec"] | None = None
     is_active: bool | None = None
     timetables: List["TimetableCreate"] | None = None
+    oneshot: OneShotAccessBase | None = None
     
 
 #### Card

app/services/door.py

@@ -4,9 +4,9 @@ from sqlmodel import select
 from fastapi import Depends, HTTPException, status
 from sqlalchemy.orm import selectinload
 import sqlalchemy.exc as exc
-import datetime
+from datetime import datetime, date, timedelta
 
-from app.services.database import Session, get_session
+from app.services.database import Session, get_session, add_and_refresh
 from app.model.models import *
 
 doorIsOpen = True
@@ -28,22 +28,36 @@ def closeDoor():
 def isDoorOpen():
     return doorIsOpen
 
+def decrementOneshot(db: Session, oneshot: OneShotAccess):
+    data = oneshot.model_dump()
+    if data["uses"] > 0:
+        data["uses"] = data["uses"] - 1
+        oneshot.sqlmodel_update(oneshot, update=data)
+        add_and_refresh(db, oneshot)
+
 def checkAccess(key: str, db: Session):
     try:
-        current_weekday = datetime.datetime.weekday(datetime.date.today())
+        current_weekday = datetime.weekday(date.today())
-        current_time = datetime.datetime.now()
+        current_time = datetime.now()
         card = db.exec(select(Card).where(Card.key == key)).one()
         for auth in card.group.accessauths:
             logger.info(f"checking auth: {auth.name}")
-            for timetable in auth.timetables:
+            if auth.type == "timetable":
-                logger.info(f"  checking timetable {timetable.id}")
+                for timetable in auth.timetables:
-                logger.info(f"   comparing weekday: CUR:{current_weekday} TT:{timetable.weekday}")
+                    logger.info(f"  checking timetable {timetable.id}")
-                if current_weekday == timetable.weekday:
+                    logger.info(f"   comparing weekday: CUR:{current_weekday} TT:{timetable.weekday}")
-                    starttime = datetime.datetime.combine(datetime.date.today(), timetable.starttime)
+                    if current_weekday == timetable.weekday:
-                    endtime = starttime + datetime.timedelta(minutes=timetable.duration)
+                        starttime = datetime.combine(date.today(), timetable.starttime)
-                    logger.info(f"   comparing time: Start:{starttime} Current:{current_time} End:{endtime}")
+                        endtime = starttime + timedelta(minutes=timetable.duration)
-                    if starttime < current_time < endtime:
+                        logger.info(f"   comparing time: Start:{starttime} Current:{current_time} End:{endtime}")
-                        logger.info("Access Valid!")
+                        if starttime < current_time < endtime:
+                            logger.info("Access Valid!")
+                            return True
+            if auth.type == "oneshot":
+                logger.info(f'  oneshot auth found: {auth.oneshot}')
+                if current_time < auth.oneshot.ends_at:
+                    if auth.oneshot.uses > 0:
+                        decrementOneshot(db, auth.oneshot)
                         return True
         logger.info("No more auths found")
         return False

test/conftest.py

@@ -3,6 +3,7 @@ from fastapi.testclient import TestClient
 from sqlmodel import Session, create_engine, SQLModel
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.pool import StaticPool
+from datetime import time
 
 from app.main import app
 from app.model.models import UserDB, Card, GroupDB, AccessAuthorizationDB, Timetable, AaGroupLink
@@ -107,11 +108,18 @@ def test_card(db_session, test_group):
 
 
 @pytest.fixture
-def test_aa(db_session):
+def test_aa_tt(db_session):
-    """Create a test access authorization."""
+    """Create a test access authorization with timetable."""
+    tt = Timetable(
+        weekday=1,
+        starttime=time(1, 0, 0, 0),
+        duration=50
+    )
     aa = AccessAuthorizationDB(
         name="Test AA",
-        is_active=True
+        is_active=True,
+        type="timetable",
+        timetables=[tt]
     )
     db_session.add(aa)
     db_session.commit()

test/test_models.py

@@ -41,7 +41,7 @@ def test_group_models():
 def test_access_authorization_models():
     """Test access authorization model creation and validation."""
     # Test AccessAuthorizationBase
-    aa_base = AccessAuthorizationBase(name="Test AA", is_active=True)
+    aa_base = AccessAuthorizationBase(name="Test AA", is_active=True, type="timetable")
     assert aa_base.name == "Test AA"
     assert aa_base.is_active is True
 
@@ -49,7 +49,8 @@ def test_access_authorization_models():
     timetable_create = TimetableCreate(weekday=1, starttime="08:00", duration=60)
     aa_create = AccessAuthorizationCreate(
         name="New AA", 
-        is_active=False, 
+        is_active=False,
+        type="timetable",
         timetables=[timetable_create]
     )
     assert aa_create.name == "New AA"

test/test_services/test_aa_manager.py

@@ -2,10 +2,11 @@ import pytest
 from fastapi import status
 
 
-def test_create_access_auth(client, auth_headers):
+def test_create_access_auth_tt(client, auth_headers):
     """Test creating a new access authorization."""
     aa_data = {
-        "name": "New AA",
+        "name": "New tt_AA",
+        "type": "timetable",
         "is_active": True,
         "timetables": [
             {"weekday": 1, "starttime": "08:00", "duration": 60},
@@ -17,13 +18,46 @@ def test_create_access_auth(client, auth_headers):
     assert response.status_code == 200
 
     data = response.json()
-    assert data["name"] == "New AA"
+    assert data["name"] == "New tt_AA"
     assert data["is_active"] is True
+    assert data["type"] == "timetable"
     assert "id" in data
     assert len(data["timetables"]) == 2
 
+def test_create_access_auth_os(client, auth_headers):
+    """Test creating a new access authorization with oneshot type."""
+    aa_data = {
+        "name": "New os_AA",
+        "type": "oneshot",
+        "is_active": True,
+        "oneshot": {
+            "uses": 1,
+            "ends_at": "2029-07-27"
+        }
+    }
 
-def test_get_all_access_auths(client, auth_headers, test_aa):
+    response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers)
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["name"] == "New os_AA"
+    assert data["type"] == "oneshot"
+    assert data["is_active"] is True
+    assert "id" in data
+    assert data["oneshot"]["uses"] == 1
+
+def test_create_wrong_aa_type(client, auth_headers):
+    """Test creating a new access authorization with oneshot type."""
+    aa_data = {
+        "name": "New AA",
+        "type": "wrong",
+        "is_active": True,
+    }
+
+    response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers)
+    assert response.status_code == 422
+
+def test_get_all_access_auths(client, auth_headers, test_aa_tt):
     """Test retrieving all access authorizations."""
     response = client.get("/api/v1/aa/", headers=auth_headers)
     assert response.status_code == 200
@@ -32,17 +66,17 @@ def test_get_all_access_auths(client, auth_headers, test_aa):
     assert len(aa_list) >= 1
 
     aa_names = [aa["name"] for aa in aa_list]
-    assert test_aa.name in aa_names
+    assert test_aa_tt.name in aa_names
 
 
-def test_get_access_auth_by_id(client, auth_headers, test_aa):
+def test_get_access_auth_by_id(client, auth_headers, test_aa_tt):
     """Test retrieving a specific access authorization by ID."""
-    response = client.get(f"/api/v1/aa/{test_aa.id}", headers=auth_headers)
+    response = client.get(f"/api/v1/aa/{test_aa_tt.id}", headers=auth_headers)
     assert response.status_code == 200
 
     data = response.json()
-    assert data["id"] == test_aa.id
+    assert data["id"] == test_aa_tt.id
-    assert data["name"] == test_aa.name
+    assert data["name"] == test_aa_tt.name
 
 
 def test_get_nonexistent_access_auth(client, auth_headers):
@@ -51,10 +85,10 @@ def test_get_nonexistent_access_auth(client, auth_headers):
     assert response.status_code == 404
 
 
-def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa):
+def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa_tt):
     """Test assigning an access authorization to a group."""
     response = client.put(
-        f"/api/v1/aa/assign/{test_group.id}/{test_aa.id}", 
+        f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", 
         headers=auth_headers
     )
     assert response.status_code == 200
@@ -65,14 +99,14 @@ def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa):
     # Note: The response model might not include the full relationship
 
 
-def test_assign_already_assigned_access_auth(client, auth_headers, test_group, test_aa):
+def test_assign_already_assigned_access_auth(client, auth_headers, test_group, test_aa_tt):
     """Test assigning an already assigned access authorization."""
     # First assignment
-    client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa.id}", headers=auth_headers)
+    client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers)
 
     # Second assignment should indicate it's already assigned
     response = client.put(
-        f"/api/v1/aa/assign/{test_group.id}/{test_aa.id}", 
+        f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", 
         headers=auth_headers
     )
     # According to the code, this returns 409 with "already assigned" message
@@ -80,31 +114,31 @@ def test_assign_already_assigned_access_auth(client, auth_headers, test_group, t
     assert "already assigned" in response.json()["detail"].lower()
 
 
-def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_aa):
+def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_aa_tt):
     """Test unassigning an access authorization from a group."""
     # First assign
-    client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa.id}", headers=auth_headers)
+    client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers)
 
     # Then unassign
     response = client.put(
-        f"/api/v1/aa/unassign/{test_group.id}/{test_aa.id}", 
+        f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", 
         headers=auth_headers
     )
     assert response.status_code == 200
 
 
-def test_unassign_nonexistent_assignment(client, auth_headers, test_group, test_aa):
+def test_unassign_nonexistent_assignment(client, auth_headers, test_group, test_aa_tt):
     """Test unassigning a non-existent assignment."""
     response = client.put(
-        f"/api/v1/aa/unassign/{test_group.id}/{test_aa.id}", 
+        f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", 
         headers=auth_headers
     )
     assert response.status_code == 404
 
 
-def test_assign_to_nonexistent_group(client, auth_headers, test_aa):
+def test_assign_to_nonexistent_group(client, auth_headers, test_aa_tt):
     """Test assigning an AA to a non-existent group."""
-    response = client.put(f"/api/v1/aa/assign/99999/{test_aa.id}", headers=auth_headers)
+    response = client.put(f"/api/v1/aa/assign/99999/{test_aa_tt.id}", headers=auth_headers)
     assert response.status_code == 404
 
 
@@ -114,7 +148,7 @@ def test_assign_nonexistent_aa(client, auth_headers, test_group):
     assert response.status_code == 404
 
 
-def test_update_access_auth(client, auth_headers, test_aa):
+def test_update_access_auth(client, auth_headers, test_aa_tt):
     """Test updating an access authorization."""
     update_data = {
         "name": "Updated AA",
@@ -122,7 +156,7 @@ def test_update_access_auth(client, auth_headers, test_aa):
     }
 
     response = client.patch(
-        f"/api/v1/aa/{test_aa.id}", 
+        f"/api/v1/aa/{test_aa_tt.id}", 
         json=update_data, 
         headers=auth_headers
     )
@@ -133,7 +167,7 @@ def test_update_access_auth(client, auth_headers, test_aa):
     assert data["is_active"] is False
 
 
-def test_update_access_auth_with_timetables(client, auth_headers, test_aa):
+def test_update_access_auth_with_timetables(client, auth_headers, test_aa_tt):
     """Test updating an access authorization with new timetables."""
     update_data = {
         "timetables": [
@@ -142,7 +176,7 @@ def test_update_access_auth_with_timetables(client, auth_headers, test_aa):
     }
 
     response = client.patch(
-        f"/api/v1/aa/{test_aa.id}", 
+        f"/api/v1/aa/{test_aa_tt.id}", 
         json=update_data, 
         headers=auth_headers
     )
@@ -161,14 +195,14 @@ def test_update_nonexistent_access_auth(client, auth_headers):
     assert response.status_code == 404
 
 
-def test_delete_access_auth(client, auth_headers, test_aa):
+def test_delete_access_auth(client, auth_headers, test_aa_tt):
     """Test deleting an access authorization."""
-    response = client.delete(f"/api/v1/aa/{test_aa.id}", headers=auth_headers)
+    response = client.delete(f"/api/v1/aa/{test_aa_tt.id}", headers=auth_headers)
     assert response.status_code == 200
     assert "deleted successfully" in response.json()["message"].lower()
 
     # Verify AA is deleted
-    response = client.get(f"/api/v1/aa/{test_aa.id}", headers=auth_headers)
+    response = client.get(f"/api/v1/aa/{test_aa_tt.id}", headers=auth_headers)
     assert response.status_code == 404
 
 
@@ -178,7 +212,7 @@ def test_delete_nonexistent_access_auth(client, auth_headers):
     assert response.status_code == 404
 
 
-def test_aa_operations_by_non_admin(client, test_aa, user_auth_headers):
+def test_aa_tt_operations_by_non_admin(client, test_aa_tt, user_auth_headers):
     """Test that non-admin users cannot perform AA operations."""
     # Try to create an AA
     response = client.post(
@@ -193,5 +227,5 @@ def test_aa_operations_by_non_admin(client, test_aa, user_auth_headers):
     assert response.status_code == 403
 
     # Try to assign AA
-    response = client.put(f"/api/v1/aa/assign/1/{test_aa.id}", headers=user_auth_headers)
+    response = client.put(f"/api/v1/aa/assign/1/{test_aa_tt.id}", headers=user_auth_headers)
     assert response.status_code == 403

test/test_services/test_door.py

@@ -1,7 +1,7 @@
 import pytest
 import datetime
 from app.services.door import checkAccess
-from app.model.models import Card, GroupDB, AccessAuthorizationDB, Timetable
+from app.model.models import Card, GroupDB, AccessAuthorizationDB, Timetable, OneShotAccess
 
 def test_check_access_with_valid_timetable(db_session):
     # Setup: create card with valid access
@@ -19,7 +19,7 @@ def test_check_access_with_valid_timetable(db_session):
     )
     db_session.add(timetable)
 
-    aa = AccessAuthorizationDB(name="Test AA", is_active=True)
+    aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable")
     db_session.add(aa)
     aa.timetables = [timetable]
     group.accessauths = [aa]
@@ -46,7 +46,7 @@ def test_check_access_outside_hours(db_session):
     )
     db_session.add(timetable)
 
-    aa = AccessAuthorizationDB(name="Test AA", is_active=True)
+    aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable")
     db_session.add(aa)
     aa.timetables = [timetable]
     group.accessauths = [aa]
@@ -55,6 +55,33 @@ def test_check_access_outside_hours(db_session):
     result = checkAccess("test-key-123", db_session)
     assert result == False
 
+def test_check_access_with_valid_oneshot(db_session):
+    # Setup: create card with valid access
+    group = GroupDB(name="Test Group")
+    db_session.add(group)
+    db_session.commit()
+
+    card = Card(key="test-key-123", group_id=group.id, enabled=True, name="test_card", card_serial="00:00:00:00:00:00:00")
+    db_session.add(card)
+
+    oneshot = OneShotAccess(
+        uses=1,
+        ends_at=datetime.datetime.now() + datetime.timedelta(days=1)
+    )
+    db_session.add(oneshot)
+
+    aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="oneshot")
+    db_session.add(aa)
+    aa.oneshot = oneshot
+    group.accessauths = [aa]
+
+    db_session.commit()
+
+    # Test: access should be granted within time window
+    result = checkAccess("test-key-123", db_session)
+    assert result == True
+    assert aa.oneshot.uses == 0
+
 def test_check_access_invalid_card(db_session):
     # Should raise exception for non-existent card
     with pytest.raises(Exception):