database: add create-owner, migrate-card, and cleanup commands

core/data/schema/card.py

@@ -3,6 +3,7 @@ from sqlalchemy import Table, Column, UniqueConstraint
 from sqlalchemy.types import Integer, String, Boolean, TIMESTAMP
 from sqlalchemy.sql.schema import ForeignKey
 from sqlalchemy.sql import func
+from sqlalchemy.engine import Row
 
 from core.data.schema.base import BaseData, metadata
 
@@ -21,21 +22,44 @@ aime_card = Table(
 )
 
 class CardData(BaseData):
-    def get_user_id_from_card(self, access_code: str) -> Optional[int]:
-        """
-        Given a 20 digit access code as a string, get the user id associated with that card
-        """
+    def get_card_by_access_code(self, access_code: str) -> Optional[Row]:
         sql = aime_card.select(aime_card.c.access_code == access_code)
 
         result = self.execute(sql)
         if result is None: return None 
+        return result.fetchone()
 
-        card = result.fetchone()
+    def get_card_by_id(self, card_id: int) -> Optional[Row]:
+        sql = aime_card.select(aime_card.c.id == card_id)
+
+        result = self.execute(sql)
+        if result is None: return None 
+        return result.fetchone()
+    
+    def update_access_code(self, old_ac: str, new_ac: str) -> None:
+        sql = aime_card.update(aime_card.c.access_code == old_ac).values(access_code = new_ac)
+
+        result = self.execute(sql)
+        if result is None: 
+            self.logger.error(f"Failed to change card access code from {old_ac} to {new_ac}")
+
+    def get_user_id_from_card(self, access_code: str) -> Optional[int]:
+        """
+        Given a 20 digit access code as a string, get the user id associated with that card
+        """
+        card = self.get_card_by_access_code(access_code)
         if card is None: return None
 
         return int(card["user"])
+    
+    def delete_card(self, card_id: int) -> None:
+        sql = aime_card.delete(aime_card.c.id == card_id)
 
-    def get_user_cards(self, aime_id: int) -> Optional[List[Dict]]:
+        result = self.execute(sql)
+        if result is None: 
+            self.logger.error(f"Failed to delete card with id {card_id}")
+
+    def get_user_cards(self, aime_id: int) -> Optional[List[Row]]:
         """
         Returns all cards owned by a user
         """

core/data/schema/user.py

@@ -1,13 +1,10 @@
 from enum import Enum
-from typing import Dict, Optional
-from sqlalchemy import Table, Column, and_
+from typing import Optional, List
+from sqlalchemy import Table, Column
 from sqlalchemy.types import Integer, String, TIMESTAMP
-from sqlalchemy.sql.schema import ForeignKey
 from sqlalchemy.sql import func
 from sqlalchemy.dialects.mysql import insert
-from sqlalchemy.sql import func, select, Delete
-from uuid import uuid4
-from datetime import datetime, timedelta
+from sqlalchemy.sql import func, select
 from sqlalchemy.engine import Row
 import bcrypt
 
@@ -34,9 +31,6 @@ class PermissionBits(Enum):
 
 class UserData(BaseData):
     def create_user(self, id: int = None, username: str = None, email: str = None, password: str = None, permission: int = 1) -> Optional[int]:
-        if email is None:
-            permission = 1
-
         if id is None:
             sql = insert(aime_user).values(
                 username=username, 
@@ -82,4 +76,22 @@ class UserData(BaseData):
     def reset_autoincrement(self, ai_value: int) -> None:
         # ALTER TABLE isn't in sqlalchemy so we do this the ugly way
         sql = f"ALTER TABLE aime_user AUTO_INCREMENT={ai_value}"
-        self.execute(sql)
+        self.execute(sql)
+    
+    def delete_user(self, user_id: int) -> None:
+        sql = aime_user.delete(aime_user.c.id == user_id)
+
+        result = self.execute(sql)
+        if result is None: 
+            self.logger.error(f"Failed to delete user with id {user_id}")
+
+    def get_unregistered_users(self) -> List[Row]:
+        """
+        Returns a list of users who have not registered with the webui. They may or may not have cards.
+        """
+        sql = select(aime_user).where(aime_user.c.password == None)
+
+        result = self.execute(sql)
+        if result is None: 
+            return None
+        return result.fetchall()