added invoke task show-users-table

src/smolvault/main.py

@@ -6,25 +6,13 @@
 from logging.handlers import RotatingFileHandler
 from typing import Annotated
 
-from fastapi import (
-    BackgroundTasks,
-    Depends,
-    FastAPI,
-    File,
-    Form,
-    HTTPException,
-    UploadFile,
-)
+from fastapi import BackgroundTasks, Depends, FastAPI, File, Form, HTTPException, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.responses import FileResponse, Response
 from fastapi.security import OAuth2PasswordRequestForm
 
-from smolvault.auth.decoder import (
-    authenticate_user,
-    create_access_token,
-    get_current_user,
-)
+from smolvault.auth.decoder import authenticate_user, create_access_token, get_current_user
 from smolvault.auth.models import NewUserDTO, Token, User
 from smolvault.cache.cache_manager import CacheManager
 from smolvault.clients.aws import S3Client

tasks.py

@@ -1,10 +1,12 @@
 import sqlite3
 from datetime import datetime
+from typing import Any
 from zoneinfo import ZoneInfo
 
 from invoke.context import Context
 from invoke.tasks import task
 from rich import print
+from rich.table import Table
 
 
 @task
@@ -36,6 +38,28 @@ def show_table(c: Context) -> None:
     conn.close()
 
 
+def output_table(title: str, column_names: list[str], rows: list[Any]) -> None:
+    table = Table(title=title)
+    for column_name in column_names:
+        table.add_column(column_name)
+    for row in rows:
+        table.add_row(*row)
+    print(table)
+
+
+@task
+def show_users_table(c: Context) -> None:
+    conn = sqlite3.connect("file_metadata.db")
+    cursor = conn.cursor()
+    cursor.execute("SELECT * FROM userinfo")
+    results = cursor.fetchall()
+    conn.close()
+    rows: list[tuple[str, str, str, str]] = []
+    for result in results:
+        rows.append((str(result[0]), result[1], result[2], result[4]))  # noqa: PERF401
+    output_table("Users Table", ["id", "username", "hashed_pwd", "name"], rows)
+
+
 @task
 def bak_db(c: Context) -> None:
     timestamp = datetime.now(ZoneInfo("UTC")).strftime("%Y-%m-%d_%H:%M:%S")