fe4cbc3537
- Web UI: login, dashboard, links CRUD, collections, API keys, admin pages - Query engine: AND/OR/XOR with field filters, tag search, preview endpoint - Session management: token expiry detection, 401 interceptor, expiry banner - Links search: tags included, multi-word AND, query mode with set operations - Collections: static/dynamic, query builder with preview, public tree view - Save as Collection: convert search results (static) or query (dynamic) - Dashboard stats: resilient loading with allSettled pattern - Login page: redesigned with public collections tree view - Bug fix: query executor None fields crash (notes/description/url/title) - E2E tests: 20 Playwright tests covering all critical user flows - All 104 tests passing (84 unit/integration + 20 E2E)
230 lines
7.5 KiB
Python
230 lines
7.5 KiB
Python
"""
|
|
LinkSyncServer - Database Base Models
|
|
"""
|
|
|
|
import os
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import (
|
|
create_engine,
|
|
Column,
|
|
Integer,
|
|
String,
|
|
Text,
|
|
DateTime,
|
|
Boolean,
|
|
ForeignKey,
|
|
JSON,
|
|
text,
|
|
)
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import relationship, sessionmaker
|
|
from sqlalchemy.sql import func
|
|
|
|
Base = declarative_base()
|
|
|
|
|
|
def get_engine():
|
|
"""Get database engine from environment variable."""
|
|
database_url = os.environ.get("DATABASE_URL", "sqlite:///linksync.db")
|
|
return create_engine(database_url, echo=False, future=True)
|
|
|
|
|
|
def get_session():
|
|
"""Get a new database session."""
|
|
engine = get_engine()
|
|
Session = sessionmaker(bind=engine)
|
|
return Session()
|
|
|
|
|
|
def init_db():
|
|
"""Initialize database tables."""
|
|
engine = get_engine()
|
|
Base.metadata.create_all(engine)
|
|
|
|
|
|
class TimestampMixin:
|
|
"""Mixin for timestamps."""
|
|
created_at = Column(
|
|
DateTime, server_default=func.now(), nullable=False
|
|
)
|
|
updated_at = Column(
|
|
DateTime, server_default=func.now(), onupdate=func.now(), nullable=False
|
|
)
|
|
|
|
|
|
class User(Base, TimestampMixin):
|
|
"""User model for authentication."""
|
|
__tablename__ = "users"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
username = Column(String(100), unique=True, nullable=False, index=True)
|
|
email = Column(String(255), unique=True, nullable=False, index=True)
|
|
password_hash = Column(String(255), nullable=False)
|
|
role = Column(String(20), nullable=False, default="user")
|
|
is_active = Column(Boolean, default=True)
|
|
|
|
bookmarks = relationship("Bookmark", back_populates="user")
|
|
collections = relationship("Collection", back_populates="user")
|
|
api_keys = relationship("ApiKey", back_populates="user")
|
|
audit_logs = relationship("AuditLog", back_populates="user")
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": str(self.id),
|
|
"username": self.username,
|
|
"email": self.email,
|
|
"role": self.role,
|
|
"is_active": self.is_active,
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
|
|
}
|
|
|
|
|
|
class ApiKey(Base, TimestampMixin):
|
|
"""API Key for authentication."""
|
|
__tablename__ = "api_keys"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False, index=True)
|
|
key_hash = Column(String(255), nullable=False, unique=True)
|
|
name = Column(String(100))
|
|
expires_at = Column(DateTime)
|
|
is_active = Column(Boolean, default=True)
|
|
|
|
user = relationship("User", back_populates="api_keys")
|
|
|
|
|
|
class Tag(Base, TimestampMixin):
|
|
"""Tag model for bookmarks."""
|
|
__tablename__ = "tags"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
name = Column(String(100), unique=True, nullable=False, index=True)
|
|
color = Column(String(7))
|
|
description = Column(Text)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": str(self.id),
|
|
"name": self.name,
|
|
"color": self.color,
|
|
"description": self.description,
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
|
|
}
|
|
|
|
|
|
class Bookmark(Base, TimestampMixin):
|
|
"""Bookmark/Link model with Firefox-compatible fields."""
|
|
__tablename__ = "links"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
url = Column(String(2048), nullable=False, index=True)
|
|
title = Column(String(255), nullable=False)
|
|
description = Column(Text)
|
|
notes = Column(Text)
|
|
tags = Column(JSON, default=list)
|
|
favicon_url = Column(String(512))
|
|
path = Column(String(512), nullable=True)
|
|
visit_count = Column(Integer, default=0)
|
|
is_bookmarked = Column(Boolean, default=False)
|
|
source_set_id = Column(UUID(as_uuid=True), ForeignKey("links.id"))
|
|
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
|
|
|
|
user = relationship("User", back_populates="bookmarks")
|
|
source_set = relationship("Bookmark", remote_side=[id])
|
|
collection_bookmarks = relationship("CollectionBookmark", back_populates="bookmark")
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": str(self.id),
|
|
"url": self.url,
|
|
"title": self.title,
|
|
"description": self.description,
|
|
"notes": self.notes,
|
|
"tags": self.tags or [],
|
|
"favicon_url": self.favicon_url,
|
|
"path": self.path,
|
|
"visit_count": self.visit_count,
|
|
"is_bookmarked": self.is_bookmarked,
|
|
"source_set_id": str(self.source_set_id) if self.source_set_id else None,
|
|
"user_id": str(self.user_id) if self.user_id else None,
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
|
|
}
|
|
|
|
|
|
class Collection(Base, TimestampMixin):
|
|
"""Collection model for bookmark sets."""
|
|
__tablename__ = "collections"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
name = Column(String(200), nullable=False)
|
|
description = Column(Text)
|
|
query_type = Column(String(20), nullable=False)
|
|
query_expression = Column(JSON)
|
|
is_public = Column(Boolean, default=False)
|
|
created_by = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
|
|
|
|
user = relationship("User", back_populates="collections")
|
|
collection_bookmarks = relationship("CollectionBookmark", back_populates="collection")
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": str(self.id),
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"query_type": self.query_type,
|
|
"query_expression": self.query_expression,
|
|
"is_public": self.is_public,
|
|
"created_by": str(self.created_by),
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
|
|
}
|
|
|
|
|
|
class CollectionBookmark(Base, TimestampMixin):
|
|
"""Junction table for static collections."""
|
|
__tablename__ = "collection_links"
|
|
|
|
collection_id = Column(UUID(as_uuid=True), ForeignKey("collections.id"), primary_key=True)
|
|
link_id = Column(UUID(as_uuid=True), ForeignKey("links.id"), primary_key=True)
|
|
|
|
collection = relationship("Collection", back_populates="collection_bookmarks")
|
|
bookmark = relationship("Bookmark", back_populates="collection_bookmarks")
|
|
|
|
|
|
class AuditLog(Base, TimestampMixin):
|
|
"""Audit log for tracking changes."""
|
|
__tablename__ = "audit_log"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
|
|
action = Column(String(100), nullable=False)
|
|
entity_type = Column(String(50), nullable=False)
|
|
entity_id = Column(UUID(as_uuid=True))
|
|
old_value = Column(JSON)
|
|
new_value = Column(JSON)
|
|
ip_address = Column(String(45))
|
|
|
|
user = relationship("User", back_populates="audit_logs")
|
|
|
|
|
|
__all__ = [
|
|
"Base",
|
|
"get_engine",
|
|
"get_session",
|
|
"init_db",
|
|
"TimestampMixin",
|
|
"User",
|
|
"ApiKey",
|
|
"Tag",
|
|
"Bookmark",
|
|
"Collection",
|
|
"CollectionBookmark",
|
|
"AuditLog",
|
|
]
|