42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
import asyncio
|
|
import getpass
|
|
|
|
import typer
|
|
from sqlalchemy import select
|
|
|
|
from app.core.security import hash_password
|
|
from app.db.session import async_session_maker
|
|
from app.models.user import User
|
|
|
|
app = typer.Typer()
|
|
|
|
|
|
@app.command()
|
|
def create_admin(username: str = typer.Option(..., prompt=True)) -> None:
|
|
"""Create an admin user for the inventory backend."""
|
|
password = getpass.getpass("Password: ")
|
|
confirm = getpass.getpass("Confirm password: ")
|
|
if password != confirm:
|
|
typer.echo("Passwords do not match", err=True)
|
|
raise typer.Exit(code=1)
|
|
if len(password) < 8:
|
|
typer.echo("Password must be at least 8 characters", err=True)
|
|
raise typer.Exit(code=1)
|
|
|
|
asyncio.run(_create_admin(username, password))
|
|
typer.echo(f"Created user '{username}'")
|
|
|
|
|
|
async def _create_admin(username: str, password: str) -> None:
|
|
async with async_session_maker() as db:
|
|
existing = await db.execute(select(User).where(User.username == username))
|
|
if existing.scalar_one_or_none() is not None:
|
|
typer.echo(f"User '{username}' already exists", err=True)
|
|
raise typer.Exit(code=1)
|
|
db.add(User(username=username, hashed_password=hash_password(password)))
|
|
await db.commit()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app()
|