Label edit functionality added
Build LinkLog Development Image / development-image (push) Successful in 10s
Build LinkLog Development Image / development-image (push) Successful in 10s
This commit is contained in:
@@ -10,7 +10,7 @@ from pydantic import BaseModel
|
||||
|
||||
from backend.app.api.dependencies import require_admin
|
||||
from backend.app.database import get_connection, hash_password
|
||||
from backend.app.services.link_service import delete_label
|
||||
from backend.app.services.link_service import delete_label, update_label
|
||||
from backend.app.services.email_service import (
|
||||
get_smtp_settings,
|
||||
save_smtp_settings,
|
||||
@@ -35,6 +35,10 @@ class AdminPluginUpdate(BaseModel):
|
||||
config: dict | None = None
|
||||
|
||||
|
||||
class AdminLabelUpdate(BaseModel):
|
||||
name: str
|
||||
|
||||
|
||||
class AdminUserCreate(BaseModel):
|
||||
username: str
|
||||
email: str
|
||||
@@ -304,6 +308,18 @@ def admin_delete_label(label_id: str, current_user: dict = Depends(require_admin
|
||||
return {'status': 'deleted', 'id': label_id}
|
||||
|
||||
|
||||
@router.put('/labels/{label_id}')
|
||||
def admin_edit_label(label_id: str, payload: AdminLabelUpdate, current_user: dict = Depends(require_admin)):
|
||||
try:
|
||||
result = update_label(label_id, user_id=current_user['id'], name=payload.name, is_admin=True)
|
||||
except ValueError as error:
|
||||
raise HTTPException(status_code=409, detail=str(error)) from error
|
||||
if result is None:
|
||||
raise HTTPException(status_code=404, detail='Label not found')
|
||||
record_audit_event(current_user['id'], 'label_updated', 'label', label_id)
|
||||
return result
|
||||
|
||||
|
||||
@router.get('/labels')
|
||||
def admin_list_labels(_: dict = Depends(require_admin)):
|
||||
with get_connection() as conn:
|
||||
|
||||
@@ -38,6 +38,22 @@ def get_current_user(
|
||||
return dict(user)
|
||||
|
||||
|
||||
def get_optional_current_user(
|
||||
credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
|
||||
) -> dict | None:
|
||||
if credentials is None or credentials.scheme.lower() != 'bearer':
|
||||
return None
|
||||
token_data = validate_token(credentials.credentials)
|
||||
if token_data is None:
|
||||
return None
|
||||
with get_connection() as conn:
|
||||
user = conn.execute(
|
||||
'SELECT * FROM users WHERE id = ?',
|
||||
(token_data['user_id'],),
|
||||
).fetchone()
|
||||
return dict(user) if user else None
|
||||
|
||||
|
||||
def require_admin(user: dict = Depends(get_current_user)):
|
||||
if not user['is_admin']:
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='Administrator access required')
|
||||
|
||||
@@ -2,10 +2,11 @@
|
||||
## SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
import json
|
||||
from fastapi import APIRouter, Header, HTTPException, Response, status
|
||||
from fastapi import APIRouter, Depends, Header, HTTPException, Response, status
|
||||
import logging
|
||||
from pydantic import BaseModel
|
||||
|
||||
from backend.app.api.dependencies import get_optional_current_user
|
||||
from backend.app.services.link_service import create_link, delete_link, find_owned_link_by_title_url, get_link_tags, get_owned_link, list_public_links, list_tags, mark_mastodon_posted, update_link
|
||||
from backend.app.database import get_connection
|
||||
from backend.app.services.plugin_manager import plugin_manager
|
||||
@@ -33,8 +34,9 @@ class LinkUpdate(BaseModel):
|
||||
|
||||
|
||||
@router.get('/tags')
|
||||
def available_tags():
|
||||
return list_tags()
|
||||
def available_tags(user: dict | None = Depends(get_optional_current_user)):
|
||||
user_id = user['id'] if user else None
|
||||
return list_tags(user_id=user_id)
|
||||
|
||||
|
||||
@router.get('/scrape')
|
||||
|
||||
@@ -27,7 +27,7 @@ def normalize_tags(tags: list[str] | None) -> list[str]:
|
||||
return normalized
|
||||
|
||||
|
||||
def save_link_tags(conn, link_id: str, tags: list[str]) -> None:
|
||||
def save_link_tags(conn, link_id: str, tags: list[str], user_id: str | None = None) -> list[str]:
|
||||
canonical_tags = []
|
||||
for tag in tags:
|
||||
tag_row = conn.execute(
|
||||
@@ -36,8 +36,8 @@ def save_link_tags(conn, link_id: str, tags: list[str]) -> None:
|
||||
).fetchone()
|
||||
if tag_row is None:
|
||||
conn.execute(
|
||||
'INSERT INTO tags (id, name) VALUES (?, ?)',
|
||||
(str(uuid4()), tag),
|
||||
'INSERT INTO tags (id, name, created_by) VALUES (?, ?, ?)',
|
||||
(str(uuid4()), tag, user_id),
|
||||
)
|
||||
tag_row = conn.execute('SELECT id FROM tags WHERE name = ?', (tag,)).fetchone()
|
||||
conn.execute(
|
||||
@@ -103,7 +103,7 @@ def create_link(
|
||||
record['is_public'],
|
||||
),
|
||||
)
|
||||
stored_tags = save_link_tags(conn, record['id'], normalized_tags)
|
||||
stored_tags = save_link_tags(conn, record['id'], normalized_tags, user_id=user_id)
|
||||
conn.commit()
|
||||
record['tags'] = stored_tags
|
||||
return record
|
||||
@@ -173,7 +173,7 @@ def update_link(
|
||||
if cursor.rowcount == 0:
|
||||
return None
|
||||
conn.execute('DELETE FROM link_tags WHERE link_id = ?', (link_id,))
|
||||
stored_tags = save_link_tags(conn, link_id, normalized_tags)
|
||||
stored_tags = save_link_tags(conn, link_id, normalized_tags, user_id=user_id)
|
||||
conn.commit()
|
||||
row = conn.execute('SELECT * FROM links WHERE id = ?', (link_id,)).fetchone()
|
||||
record = dict(row)
|
||||
@@ -227,9 +227,32 @@ def list_public_users():
|
||||
return [row['username'] for row in rows]
|
||||
|
||||
|
||||
def list_tags():
|
||||
def list_tags(user_id: str | None = None):
|
||||
with get_connection() as conn:
|
||||
rows = conn.execute('SELECT name FROM tags ORDER BY name').fetchall()
|
||||
if user_id:
|
||||
rows = conn.execute(
|
||||
'''
|
||||
SELECT tags.name
|
||||
FROM tags
|
||||
LEFT JOIN users ON users.id = tags.created_by
|
||||
WHERE tags.created_by IS NULL
|
||||
OR tags.created_by = ?
|
||||
OR users.is_admin = 1
|
||||
ORDER BY tags.name
|
||||
''',
|
||||
(user_id,),
|
||||
).fetchall()
|
||||
else:
|
||||
rows = conn.execute(
|
||||
'''
|
||||
SELECT tags.name
|
||||
FROM tags
|
||||
LEFT JOIN users ON users.id = tags.created_by
|
||||
WHERE tags.created_by IS NULL
|
||||
OR users.is_admin = 1
|
||||
ORDER BY tags.name
|
||||
''',
|
||||
).fetchall()
|
||||
tags = []
|
||||
seen = set()
|
||||
for row in rows:
|
||||
@@ -242,7 +265,15 @@ def list_tags():
|
||||
def list_user_labels(user_id: str):
|
||||
with get_connection() as conn:
|
||||
rows = conn.execute(
|
||||
'SELECT id, name, created_by FROM tags WHERE created_by = ? ORDER BY name',
|
||||
'''
|
||||
SELECT tags.id, tags.name, tags.created_by, users.username AS creator
|
||||
FROM tags
|
||||
LEFT JOIN users ON users.id = tags.created_by
|
||||
WHERE tags.created_by IS NULL
|
||||
OR tags.created_by = ?
|
||||
OR users.is_admin = 1
|
||||
ORDER BY tags.name
|
||||
''',
|
||||
(user_id,),
|
||||
).fetchall()
|
||||
return [dict(row) for row in rows]
|
||||
@@ -268,7 +299,7 @@ def create_label(user_id: str, name: str):
|
||||
return {'id': label_id, 'name': label, 'created_by': user_id}
|
||||
|
||||
|
||||
def update_label(label_id: str, user_id: str, name: str):
|
||||
def update_label(label_id: str, user_id: str | None = None, name: str = '', is_admin: bool = False):
|
||||
normalized = normalize_tags([name])
|
||||
if not normalized:
|
||||
raise ValueError('Label cannot be empty')
|
||||
@@ -276,7 +307,7 @@ def update_label(label_id: str, user_id: str, name: str):
|
||||
current = conn.execute(
|
||||
'SELECT id, name, created_by FROM tags WHERE id = ?', (label_id,)
|
||||
).fetchone()
|
||||
if current is None or current['created_by'] != user_id:
|
||||
if current is None or (not is_admin and current['created_by'] != user_id):
|
||||
return None
|
||||
duplicate = conn.execute(
|
||||
'SELECT id FROM tags WHERE lower(name) = lower(?) AND id != ?',
|
||||
@@ -286,7 +317,7 @@ def update_label(label_id: str, user_id: str, name: str):
|
||||
raise ValueError('Label already exists')
|
||||
conn.execute('UPDATE tags SET name = ? WHERE id = ?', (normalized[0], label_id))
|
||||
conn.commit()
|
||||
return {'id': label_id, 'name': normalized[0], 'created_by': user_id}
|
||||
return {'id': label_id, 'name': normalized[0], 'created_by': current['created_by']}
|
||||
|
||||
|
||||
def delete_label(label_id: str, user_id: str | None = None, is_admin: bool = False):
|
||||
|
||||
Reference in New Issue
Block a user