Add admin user

This commit is contained in:
dvirlabs 2025-12-08 09:24:56 +02:00
parent fa5ba578bb
commit 8d81d16682
6 changed files with 23 additions and 13 deletions

View File

@ -94,6 +94,7 @@ class UserResponse(BaseModel):
first_name: Optional[str] = None first_name: Optional[str] = None
last_name: Optional[str] = None last_name: Optional[str] = None
display_name: str display_name: str
is_admin: bool = False
app = FastAPI( app = FastAPI(
@ -164,7 +165,7 @@ def create_recipe(recipe_in: RecipeCreate, current_user: dict = Depends(get_curr
@app.put("/recipes/{recipe_id}", response_model=Recipe) @app.put("/recipes/{recipe_id}", response_model=Recipe)
def update_recipe(recipe_id: int, recipe_in: RecipeUpdate, current_user: dict = Depends(get_current_user)): def update_recipe(recipe_id: int, recipe_in: RecipeUpdate, current_user: dict = Depends(get_current_user)):
# Check ownership BEFORE updating # Check ownership BEFORE updating (admins can edit any recipe)
conn = get_conn() conn = get_conn()
try: try:
with conn.cursor() as cur: with conn.cursor() as cur:
@ -172,7 +173,8 @@ def update_recipe(recipe_id: int, recipe_in: RecipeUpdate, current_user: dict =
recipe = cur.fetchone() recipe = cur.fetchone()
if not recipe: if not recipe:
raise HTTPException(status_code=404, detail="המתכון לא נמצא") raise HTTPException(status_code=404, detail="המתכון לא נמצא")
if recipe["user_id"] != current_user["user_id"]: # Allow if user is owner OR admin
if recipe["user_id"] != current_user["user_id"] and not current_user.get("is_admin", False):
raise HTTPException(status_code=403, detail="אין לך הרשאה לערוך מתכון זה") raise HTTPException(status_code=403, detail="אין לך הרשאה לערוך מתכון זה")
finally: finally:
conn.close() conn.close()
@ -201,7 +203,7 @@ def update_recipe(recipe_id: int, recipe_in: RecipeUpdate, current_user: dict =
@app.delete("/recipes/{recipe_id}", status_code=204) @app.delete("/recipes/{recipe_id}", status_code=204)
def delete_recipe(recipe_id: int, current_user: dict = Depends(get_current_user)): def delete_recipe(recipe_id: int, current_user: dict = Depends(get_current_user)):
# Get recipe first to check ownership # Get recipe first to check ownership (admins can delete any recipe)
conn = get_conn() conn = get_conn()
try: try:
with conn.cursor() as cur: with conn.cursor() as cur:
@ -209,7 +211,8 @@ def delete_recipe(recipe_id: int, current_user: dict = Depends(get_current_user)
recipe = cur.fetchone() recipe = cur.fetchone()
if not recipe: if not recipe:
raise HTTPException(status_code=404, detail="המתכון לא נמצא") raise HTTPException(status_code=404, detail="המתכון לא נמצא")
if recipe["user_id"] != current_user["user_id"]: # Allow if user is owner OR admin
if recipe["user_id"] != current_user["user_id"] and not current_user.get("is_admin", False):
raise HTTPException(status_code=403, detail="אין לך הרשאה למחוק מתכון זה") raise HTTPException(status_code=403, detail="אין לך הרשאה למחוק מתכון זה")
finally: finally:
conn.close() conn.close()

View File

@ -7,6 +7,7 @@ CREATE TABLE IF NOT EXISTS users (
first_name TEXT, first_name TEXT,
last_name TEXT, last_name TEXT,
display_name TEXT UNIQUE NOT NULL, display_name TEXT UNIQUE NOT NULL,
is_admin BOOLEAN DEFAULT FALSE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
); );
@ -41,4 +42,10 @@ CREATE INDEX IF NOT EXISTS idx_recipes_made_by
CREATE INDEX IF NOT EXISTS idx_recipes_user_id CREATE INDEX IF NOT EXISTS idx_recipes_user_id
ON recipes (user_id); ON recipes (user_id);
-- Create default admin user (password: admin123)
-- Password hash generated with bcrypt for 'admin123'
INSERT INTO users (username, email, password_hash, first_name, last_name, display_name, is_admin)
VALUES ('admin', 'admin@myrecipes.local', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/LewY5lE7UGf3rCvHC', 'Admin', 'User', 'מנהל', TRUE)
ON CONFLICT (username) DO NOTHING;

View File

@ -14,7 +14,7 @@ def get_db_connection():
) )
def create_user(username: str, email: str, password_hash: str, first_name: str = None, last_name: str = None, display_name: str = None): def create_user(username: str, email: str, password_hash: str, first_name: str = None, last_name: str = None, display_name: str = None, is_admin: bool = False):
"""Create a new user""" """Create a new user"""
conn = get_db_connection() conn = get_db_connection()
cur = conn.cursor(cursor_factory=RealDictCursor) cur = conn.cursor(cursor_factory=RealDictCursor)
@ -24,11 +24,11 @@ def create_user(username: str, email: str, password_hash: str, first_name: str =
cur.execute( cur.execute(
""" """
INSERT INTO users (username, email, password_hash, first_name, last_name, display_name) INSERT INTO users (username, email, password_hash, first_name, last_name, display_name, is_admin)
VALUES (%s, %s, %s, %s, %s, %s) VALUES (%s, %s, %s, %s, %s, %s, %s)
RETURNING id, username, email, first_name, last_name, display_name, created_at RETURNING id, username, email, first_name, last_name, display_name, is_admin, created_at
""", """,
(username, email, password_hash, first_name, last_name, final_display_name) (username, email, password_hash, first_name, last_name, final_display_name, is_admin)
) )
user = cur.fetchone() user = cur.fetchone()
conn.commit() conn.commit()
@ -44,7 +44,7 @@ def get_user_by_username(username: str):
cur = conn.cursor(cursor_factory=RealDictCursor) cur = conn.cursor(cursor_factory=RealDictCursor)
try: try:
cur.execute( cur.execute(
"SELECT id, username, email, password_hash, first_name, last_name, display_name, created_at FROM users WHERE username = %s", "SELECT id, username, email, password_hash, first_name, last_name, display_name, is_admin, created_at FROM users WHERE username = %s",
(username,) (username,)
) )
user = cur.fetchone() user = cur.fetchone()
@ -60,7 +60,7 @@ def get_user_by_email(email: str):
cur = conn.cursor(cursor_factory=RealDictCursor) cur = conn.cursor(cursor_factory=RealDictCursor)
try: try:
cur.execute( cur.execute(
"SELECT id, username, email, password_hash, first_name, last_name, display_name, created_at FROM users WHERE email = %s", "SELECT id, username, email, password_hash, first_name, last_name, display_name, is_admin, created_at FROM users WHERE email = %s",
(email,) (email,)
) )
user = cur.fetchone() user = cur.fetchone()
@ -76,7 +76,7 @@ def get_user_by_id(user_id: int):
cur = conn.cursor(cursor_factory=RealDictCursor) cur = conn.cursor(cursor_factory=RealDictCursor)
try: try:
cur.execute( cur.execute(
"SELECT id, username, email, first_name, last_name, display_name, created_at FROM users WHERE id = %s", "SELECT id, username, email, first_name, last_name, display_name, is_admin, created_at FROM users WHERE id = %s",
(user_id,) (user_id,)
) )
user = cur.fetchone() user = cur.fetchone()
@ -92,7 +92,7 @@ def get_user_by_display_name(display_name: str):
cur = conn.cursor(cursor_factory=RealDictCursor) cur = conn.cursor(cursor_factory=RealDictCursor)
try: try:
cur.execute( cur.execute(
"SELECT id, username, email, display_name, created_at FROM users WHERE display_name = %s", "SELECT id, username, email, display_name, is_admin, created_at FROM users WHERE display_name = %s",
(display_name,) (display_name,)
) )
user = cur.fetchone() user = cur.fetchone()