updated

fastapi_auth/auth_config.py

@@ -3,12 +3,12 @@
 import os
 
 DEFAULT_KEY = "secret_key"
-DEFAULT_DB = os.path.abspath(os.path.join(os.getcwd(), "user_management.db"))
+DEFAULT_DB_URI = "sqlite:///user_management.db"
 
 
 class AuthConfig(BaseSettings):
     secret_key: SecretStr = DEFAULT_KEY
-    db_name: str = DEFAULT_DB
+    db_uri: str = DEFAULT_DB_URI
 
     class Config:
         env_prefix = "AUTH_"

fastapi_auth/auth_routes.py

@@ -8,7 +8,7 @@
 from .dependencies import get_user
 from .logger import LOG
 from .users import User, ANONYMOUS_USER
-from .user_management import UserManagement
+from .user_management_sqlobject import UserManagement
 from .jinja2_templates import templates
 from .roles import ROLES_REGISTRY
 from datetime import datetime, timezone
@@ -62,7 +62,7 @@ async def login_post(
     password: str = Form(...),
     user: User = Depends(get_user),
 ):
-    um = UserManagement(AUTH_SETTINGS.db_name)
+    um = UserManagement(AUTH_SETTINGS.db_uri)
     user_data = um.get_user(username, password)
 
     if user_data is not None:

fastapi_auth/demo_app.py

@@ -79,7 +79,12 @@ def read_root(
     return templates.TemplateResponse(
         request,
         "demo.html",
-        {"request": request, "user": user, "message": message, "error_message": error_message, },
+        {
+            "request": request,
+            "user": user,
+            "message": message,
+            "error_message": error_message,
+        },
     )
 
 

fastapi_auth/tests/conftest.py

@@ -1,7 +1,9 @@
 from fastapi.testclient import TestClient
 import pytest
-from ..user_management import UserManagement
+from sqlmodel import SQLModel, Field, Session, create_engine, select
+from ..user_management_sqlobject import UserManagement
 import tempfile
+import uuid
 
 from ..demo_app import app
 
@@ -13,19 +15,19 @@
 @pytest.fixture(scope="module")
 def user_management():
     # Ensure that the user management instance is initialized with a temporary database
-    def my_init(self, db_name):
-        self.db_filename = temp_db_filename
+    def my_init(self, db_uri):
+        self.engine = create_engine(tmp_db_uri)
+        SQLModel.metadata.create_all(self.engine)
 
     old_init = UserManagement.__init__
     UserManagement.__init__ = my_init
 
-    temp_db_filename = tempfile.mktemp(suffix=".db")
-    um = UserManagement(temp_db_filename)
-    um.create_db()
+    tmp_db = str(uuid.uuid4()) + ".db"
+    tmp_db_uri = f"sqlite:///{tmp_db}"
+    um = UserManagement(tmp_db)
     um.add_user("admin", "admin", "Administrator")
     yield um
     UserManagement.__init__ = old_init
-    um.delete_db()
 
 
 @pytest.fixture(scope="module")

fastapi_auth/user_cmd.py

@@ -4,24 +4,28 @@
 
 from .logger import LOG
 from .auth_config import AUTH_SETTINGS
+from rich.table import Table
+from rich.console import Console
 
-from .user_management import UserManagement
+
+# from .user_management import UserManagement
+from .user_management_sqlobject import UserManagement
 
 
 app = typer.Typer()
 
 
 def get_user_management() -> UserManagement:
     """Get a UserManagement instance."""
-    LOG.debug(f"Using database file {AUTH_SETTINGS.db_name}")
-    return UserManagement(AUTH_SETTINGS.db_name)
+    LOG.debug(f"Using database {AUTH_SETTINGS.db_uri}")
+    return UserManagement(AUTH_SETTINGS.db_uri)
 
 
 @app.command()
 def add(username: str, password: str, roles: str) -> None:
     """Add a user to the database."""
     um = get_user_management()
-    um.create_db()
+    #    um.create_db()
 
     roles_lst = roles.split(",")
     #    allowed_roles = ROLES_REGISTRY.all_role_names()
@@ -64,8 +68,15 @@ def list_users() -> None:
     """List all users in the database."""
     um = get_user_management()
     users = um.get_users()
+
+    console = Console()
+    table = Table(show_header=True, header_style="bold magenta")
+    table.add_column("Name")
+    table.add_column("Role")
+    table.add_column("Created")
     for user in users:
-        typer.echo(user)
+        table.add_row(user.username, user.roles, user.created.isoformat())
+    console.print(table)
 
 
 @app.command()

fastapi_auth/user_management_sqlobject.py

@@ -3,6 +3,8 @@
 from sqlmodel import SQLModel, Field, Session, create_engine, select
 from datetime import datetime, timezone
 import bcrypt
+from rich.table import Table
+from rich.console import Console
 
 
 def utc_now():
@@ -22,6 +24,8 @@ def __init__(self, db_uri) -> None:
         SQLModel.metadata.create_all(self.engine)
 
     def add_user(self, username: str, password: str, roles: str) -> None:
+        if self.has_user(username):
+            raise ValueError(f"User {username} already exists.")
         hashed_password = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
         user = User(username=username, password=hashed_password, roles=roles)
         with Session(self.engine) as session:

pyproject.toml

@@ -21,6 +21,7 @@ dependencies = [
     "bcrypt",
     "jinja2",
     "sqlmodel", 
+    "rich",
     "python-multipart",
     "typer",
     "typeguard",