# retoor <retoor@molodetz.nl>
import dataset
import logging
from pathlib import Path
from sqlalchemy import or_
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from devplacepy.cache import TTLCache
from devplacepy.config import DATABASE_URL, INTERNAL_GATEWAY_URL, ensure_data_dirs
logger = logging.getLogger(__name__)
ensure_data_dirs()
if DATABASE_URL.startswith("sqlite:///"):
_db_file = DATABASE_URL[len("sqlite:///") :]
if _db_file and _db_file != ":memory:":
Path(_db_file).parent.mkdir(parents=True, exist_ok=True)
db = dataset.connect(
DATABASE_URL,
engine_kwargs={
"connect_args": {
"timeout": 30,
"check_same_thread": False,
},
},
on_connect_statements=[
"PRAGMA journal_mode=WAL",
"PRAGMA synchronous=NORMAL",
"PRAGMA busy_timeout=30000",
"PRAGMA cache_size=-8000",
"PRAGMA temp_store=MEMORY",
"PRAGMA mmap_size=268435456",
],
)
def refresh_snapshot() -> None:
connection = db.executable
if connection.in_transaction() and not db.in_transaction:
connection.commit()
_local_cache_versions: dict = {}
_cache_version_cache = TTLCache(ttl=1)
_cache_state_ready = False
def _ensure_cache_state() -> None:
global _cache_state_ready
if _cache_state_ready:
return
db.query(
"CREATE TABLE IF NOT EXISTS cache_state "
"(name TEXT PRIMARY KEY, version INTEGER NOT NULL DEFAULT 0)"
)
_cache_state_ready = True
def get_cache_version(name: str) -> int:
cached = _cache_version_cache.get(name)
if cached is not None:
return cached
try:
_ensure_cache_state()
row = next(
iter(
db.query(
"SELECT version FROM cache_state WHERE name = :name", name=name
)
),
None,
)
version = int(row["version"]) if row else 0
except Exception as e:
logger.warning(f"Could not read cache version {name}: {e}")
return 0
_cache_version_cache.set(name, version)
return version
def bump_cache_version(name: str) -> None:
try:
_ensure_cache_state()
db.query(
"INSERT OR IGNORE INTO cache_state (name, version) VALUES (:name, 0)",
name=name,
)
db.query(
"UPDATE cache_state SET version = version + 1 WHERE name = :name", name=name
)
connection = db.executable
if connection.in_transaction():
connection.commit()
_cache_version_cache.pop(name)
except Exception as e:
logger.warning(f"Could not bump cache version {name}: {e}")
def sync_local_cache(name: str, cache) -> None:
current = get_cache_version(name)
if name not in _local_cache_versions:
_local_cache_versions[name] = current
return
if _local_cache_versions[name] != current:
cache.clear()
_local_cache_versions[name] = current
def _index(db, table, name, columns):
try:
if table in db.tables:
cols = ", ".join(columns)
db.query(f"CREATE INDEX IF NOT EXISTS {name} ON {table} ({cols})")
except Exception as e:
logger.warning(f"Could not create index {name} on {table}: {e}")
SOFT_DELETE_TABLES = [
"posts",
"comments",
"gists",
"projects",
"news",
"news_images",
"project_files",
"attachments",
"votes",
"reactions",
"bookmarks",
"follows",
"poll_votes",
"polls",
"poll_options",
"sessions",
"instances",
"instance_schedules",
"devii_conversations",
"devii_tasks",
"devii_lessons",
"devii_virtual_tools",
"user_customizations",
"project_forks",
"issue_tickets",
"issue_comment_authors",
"notification_preferences",
"deepsearch_sessions",
"deepsearch_messages",
"devrant_tokens",
]
def ensure_soft_delete_columns(table, *, db_handle=None):
handle = db_handle or db
if table not in handle.tables:
return
target = handle[table]
if not target.has_column("deleted_at"):
target.create_column_by_example("deleted_at", "")
if not target.has_column("deleted_by"):
target.create_column_by_example("deleted_by", "")
_index(handle, table, f"idx_{table}_deleted", ["deleted_at"])
BUG_TABLE_RENAMES = (
("bug_tickets", "issue_tickets"),
("bug_comment_authors", "issue_comment_authors"),
)
def migrate_bug_tables_to_issue_tables() -> None:
for source_name, destination_name in BUG_TABLE_RENAMES:
if source_name not in db.tables:
continue
source = db[source_name]
destination = get_table(destination_name)
copied = 0
for row in source.all():
payload = {key: value for key, value in row.items() if key != "id"}
destination.insert_ignore(payload, ["uid"])
copied += 1
logger.info(
"Migrated %s rows from %s into %s", copied, source_name, destination_name
)
source.drop()
logger.info("Dropped table %s after migration", source_name)
def init_db():
tables = db.tables
_index(db, "users", "idx_users_username", ["username"])
_index(db, "users", "idx_users_email", ["email"])
_index(db, "users", "idx_users_api_key", ["api_key"])
_index(db, "posts", "idx_posts_user_uid", ["user_uid"])
_index(db, "posts", "idx_posts_created_at", ["created_at"])
_index(db, "posts", "idx_posts_topic", ["topic"])
if "posts" in tables:
posts_table = get_table("posts")
if not posts_table.has_column("tags"):
posts_table.create_column_by_example("tags", "")
if "devrant_tokens" in tables:
devrant_tokens = get_table("devrant_tokens")
for column, example in (
("uid", ""),
("key", ""),
("user_uid", ""),
("user_id", 0),
("expire_time", 0),
("created_at", ""),
):
if not devrant_tokens.has_column(column):
devrant_tokens.create_column_by_example(column, example)
_index(db, "devrant_tokens", "idx_devrant_tokens_key", ["key"])
_index(db, "devrant_tokens", "idx_devrant_tokens_user", ["user_uid"])
_index(db, "comments", "idx_comments_post_uid", ["post_uid"])
_index(db, "comments", "idx_comments_target", ["target_type", "target_uid"])
_index(db, "comments", "idx_comments_user_uid", ["user_uid"])
_index(db, "comments", "idx_comments_created_at", ["created_at"])
_index(db, "votes", "idx_votes_target", ["target_uid", "target_type"])
_index(db, "messages", "idx_messages_sender", ["sender_uid"])
_index(db, "messages", "idx_messages_receiver", ["receiver_uid"])
_index(db, "notifications", "idx_notifications_user", ["user_uid"])
_index(db, "notifications", "idx_notifications_user_read", ["user_uid", "read"])
_index(db, "push_registration", "idx_push_registration_user", ["user_uid"])
_index(db, "sessions", "idx_sessions_token", ["session_token"])
_index(db, "projects", "idx_projects_user", ["user_uid"])
_index(db, "projects", "idx_projects_private", ["is_private"])
_index(db, "project_files", "idx_project_files_path", ["project_uid", "path"])
_index(
db, "project_files", "idx_project_files_parent", ["project_uid", "parent_path"]
)
_index(db, "badges", "idx_badges_user", ["user_uid"])
_index(db, "follows", "idx_follows_follower", ["follower_uid"])
_index(db, "follows", "idx_follows_following", ["following_uid"])
_index(db, "password_resets", "idx_password_resets_token", ["token"])
_index(db, "gists", "idx_gists_user_uid", ["user_uid"])
_index(db, "gists", "idx_gists_language", ["language"])
attachments = get_table("attachments")
for column, example in (
("uid", ""),
("target_type", ""),
("target_uid", ""),
("resource_type", ""),
("resource_uid", ""),
("user_uid", ""),
("original_filename", ""),
("stored_name", ""),
("directory", ""),
("file_size", 0),
("mime_type", ""),
("image_width", 0),
("image_height", 0),
("has_thumbnail", 0),
("thumbnail_name", ""),
("created_at", ""),
("deleted_at", ""),
):
if not attachments.has_column(column):
attachments.create_column_by_example(column, example)
_index(
db, "attachments", "idx_attachments_resource", ["resource_type", "resource_uid"]
)
_index(db, "attachments", "idx_attachments_target", ["target_type", "target_uid"])
_index(db, "attachments", "idx_attachments_user_created", ["user_uid", "created_at"])
_index(db, "reactions", "idx_reactions_target", ["target_type", "target_uid"])
_index(
db,
"reactions",
"idx_reactions_user_target",
["user_uid", "target_type", "target_uid"],
)
_index(db, "bookmarks", "idx_bookmarks_user", ["user_uid"])
_index(db, "bookmarks", "idx_bookmarks_target", ["target_type", "target_uid"])
_index(db, "polls", "idx_polls_post", ["post_uid"])
_index(db, "poll_options", "idx_poll_options_poll", ["poll_uid"])
_index(db, "poll_votes", "idx_poll_votes_poll", ["poll_uid"])
_index(db, "poll_votes", "idx_poll_votes_user", ["poll_uid", "user_uid"])
if "site_settings" in tables:
defaults = {
"site_name": "DevPlace",
"site_description": "The Developer Social Network",
"site_tagline": "Track industry shifts. Discover bold releases. Share what you are building in an open, uncensored environment.",
}
for key, value in defaults.items():
existing = db["site_settings"].find_one(key=key)
if not existing:
db["site_settings"].insert(
{"uid": f"default_{key}", "key": key, "value": value}
)
news = get_table("news")
for column, example in (
("uid", ""),
("slug", ""),
("title", ""),
("external_id", ""),
("status", ""),
("source_name", ""),
("url", ""),
("description", ""),
("content", ""),
("synced_at", ""),
("grade", 0),
("show_on_landing", 0),
):
if not news.has_column(column):
news.create_column_by_example(column, example)
_index(db, "news", "idx_news_external_id", ["external_id"])
_index(db, "news", "idx_news_synced_at", ["synced_at"])
_index(db, "news", "idx_news_status", ["status"])
news_images = get_table("news_images")
if not news_images.has_column("news_uid"):
news_images.create_column_by_example("news_uid", "")
if not news_images.has_column("url"):
news_images.create_column_by_example("url", "")
news_sync = get_table("news_sync")
if not news_sync.has_column("external_id"):
news_sync.create_column_by_example("external_id", "")
_index(db, "news_images", "idx_news_images_news_uid", ["news_uid"])
_index(db, "news_sync", "idx_news_sync_external_id", ["external_id"])
issue_tickets = get_table("issue_tickets")
for column, example in (
("uid", ""),
("gitea_number", 0),
("author_uid", ""),
("original_title", ""),
("original_description", ""),
("enhanced_title", ""),
("html_url", ""),
("last_status", ""),
("last_comment_count", 0),
("created_at", ""),
("updated_at", ""),
("deleted_at", ""),
("deleted_by", ""),
):
if not issue_tickets.has_column(column):
issue_tickets.create_column_by_example(column, example)
issue_comment_authors = get_table("issue_comment_authors")
for column, example in (
("uid", ""),
("gitea_comment_id", 0),
("gitea_number", 0),
("author_uid", ""),
("created_at", ""),
("deleted_at", ""),
("deleted_by", ""),
):
if not issue_comment_authors.has_column(column):
issue_comment_authors.create_column_by_example(column, example)
_index(db, "issue_tickets", "idx_issue_tickets_number", ["gitea_number"])
_index(db, "issue_tickets", "idx_issue_tickets_author", ["author_uid"])
_index(
db,
"issue_comment_authors",
"idx_issue_comment_authors_comment",
["gitea_comment_id"],
)
_index(
db, "issue_comment_authors", "idx_issue_comment_authors_number", ["gitea_number"]
)
migrate_bug_tables_to_issue_tables()
_index(db, "service_state", "idx_service_state_name", ["name"])
if "devii_conversations" in db.tables:
conversations = get_table("devii_conversations")
if not conversations.has_column("channel"):
conversations.create_column_by_example("channel", "main")
try:
db.query(
"UPDATE devii_conversations SET channel='main' WHERE channel IS NULL"
)
except Exception as e: # noqa: BLE001
logger.warning(f"Could not backfill devii_conversations.channel: {e}")
_index(
db,
"devii_conversations",
"idx_devii_conv_owner",
["owner_kind", "owner_id", "channel"],
)
_index(
db,
"devii_usage_ledger",
"idx_devii_ledger_owner_time",
["owner_kind", "owner_id", "created_at"],
)
_index(
db,
"devii_turns",
"idx_devii_turns_owner_time",
["owner_kind", "owner_id", "started_at"],
)
_index(db, "devii_tasks", "idx_devii_tasks_owner", ["owner_kind", "owner_id"])
_index(
db, "devii_tasks", "idx_devii_tasks_due", ["enabled", "status", "next_run_at"]
)
_index(db, "devii_lessons", "idx_devii_lessons_owner", ["owner_kind", "owner_id"])
_index(
db, "devii_virtual_tools", "idx_devii_vtools_owner", ["owner_kind", "owner_id"]
)
_index(
db,
"devii_virtual_tools",
"idx_devii_vtools_name",
["owner_kind", "owner_id", "name"],
)
_index(db, "devii_behavior", "idx_devii_behavior_owner", ["owner_kind", "owner_id"])
_index(db, "user_customizations", "idx_user_cust_owner", ["owner_kind", "owner_id"])
_index(
db,
"user_customizations",
"idx_user_cust_scope",
["owner_kind", "owner_id", "scope", "lang"],
)
_index(db, "gateway_usage_ledger", "idx_gw_usage_time", ["created_at"])
_index(
db,
"gateway_usage_ledger",
"idx_gw_usage_backend_time",
["backend", "created_at"],
)
_index(
db, "gateway_usage_ledger", "idx_gw_usage_model_time", ["model", "created_at"]
)
_index(
db,
"gateway_usage_ledger",
"idx_gw_usage_owner_time",
["owner_kind", "owner_id", "created_at"],
)
_index(
db,
"gateway_usage_ledger",
"idx_gw_usage_endpoint_time",
["endpoint", "created_at"],
)
_index(db, "gateway_concurrency_samples", "idx_gw_conc_time", ["created_at"])
_index(db, "jobs", "idx_jobs_kind_status", ["kind", "status"])
_index(db, "jobs", "idx_jobs_owner", ["owner_kind", "owner_id"])
_index(db, "jobs", "idx_jobs_expires", ["expires_at"])
_index(db, "project_forks", "idx_project_forks_source", ["source_project_uid"])
_index(db, "project_forks", "idx_project_forks_forked", ["forked_project_uid"])
if "instances" in db.tables:
instances = get_table("instances")
for column, example in (
("run_as_uid", ""),
("boot_language", "none"),
("boot_script", ""),
("start_on_boot", 0),
):
if not instances.has_column(column):
instances.create_column_by_example(column, example)
_index(db, "instances", "idx_instances_project", ["project_uid"])
_index(db, "instances", "idx_instances_state", ["desired_state", "status"])
_index(db, "instances", "idx_instances_container", ["container_id"])
_index(db, "instances", "idx_instances_ingress", ["ingress_slug"])
_index(
db,
"instance_events",
"idx_instance_events_instance",
["instance_uid", "created_at"],
)
_index(
db,
"instance_metrics",
"idx_instance_metrics_instance_ts",
["instance_uid", "ts"],
)
_index(
db,
"instance_schedules",
"idx_instance_schedules_due",
["enabled", "next_run_at"],
)
from devplacepy.services.audit import store as audit_store
audit_store.ensure_tables()
_index(db, "audit_log", "idx_audit_created_at", ["created_at"])
_index(db, "audit_log", "idx_audit_event_key", ["event_key"])
_index(db, "audit_log", "idx_audit_category", ["category"])
_index(db, "audit_log", "idx_audit_actor_uid", ["actor_uid"])
_index(db, "audit_log", "idx_audit_actor_role", ["actor_role"])
_index(db, "audit_log", "idx_audit_origin", ["origin"])
_index(db, "audit_log", "idx_audit_target", ["target_type", "target_uid"])
_index(db, "audit_log", "idx_audit_result", ["result"])
_index(db, "audit_log_links", "idx_audit_links_audit", ["audit_uid"])
_index(db, "audit_log_links", "idx_audit_links_object", ["object_type", "object_uid"])
_index(db, "audit_log_links", "idx_audit_links_relation", ["relation"])
notification_preferences = get_table("notification_preferences")
for column, example in (
("uid", ""),
("user_uid", ""),
("notification_type", ""),
("in_app_enabled", 1),
("push_enabled", 1),
("created_at", ""),
("updated_at", ""),
("deleted_at", ""),
("deleted_by", ""),
):
if not notification_preferences.has_column(column):
notification_preferences.create_column_by_example(column, example)
_index(db, "notification_preferences", "idx_notif_prefs_user", ["user_uid"])
_index(
db,
"notification_preferences",
"idx_notif_prefs_user_type",
["user_uid", "notification_type"],
)
deepsearch_sessions = get_table("deepsearch_sessions")
for column, example in (
("uid", ""),
("owner_kind", ""),
("owner_id", ""),
("query", ""),
("status", ""),
("depth", 0),
("max_pages", 0),
("score", 0),
("confidence", 0.0),
("source_diversity", 0.0),
("page_count", 0),
("chunk_count", 0),
("collection", ""),
("summary", ""),
("created_at", ""),
("completed_at", ""),
("deleted_at", ""),
("deleted_by", ""),
):
if not deepsearch_sessions.has_column(column):
deepsearch_sessions.create_column_by_example(column, example)
_index(db, "deepsearch_sessions", "idx_deepsearch_sessions_owner", ["owner_kind", "owner_id"])
_index(db, "deepsearch_sessions", "idx_deepsearch_sessions_status", ["status"])
_index(db, "deepsearch_sessions", "idx_deepsearch_sessions_created", ["created_at"])
deepsearch_messages = get_table("deepsearch_messages")
for column, example in (
("uid", ""),
("session_uid", ""),
("role", ""),
("content", ""),
("citations", ""),
("created_at", ""),
("deleted_at", ""),
("deleted_by", ""),
):
if not deepsearch_messages.has_column(column):
deepsearch_messages.create_column_by_example(column, example)
_index(db, "deepsearch_messages", "idx_deepsearch_messages_session", ["session_uid"])
deepsearch_url_cache = get_table("deepsearch_url_cache")
for column, example in (
("uid", ""),
("url_hash", ""),
("url", ""),
("title", ""),
("content_hash", ""),
("status", 0),
("byte_size", 0),
("fetched_at", ""),
):
if not deepsearch_url_cache.has_column(column):
deepsearch_url_cache.create_column_by_example(column, example)
_index(db, "deepsearch_url_cache", "idx_deepsearch_url_cache_hash", ["url_hash"])
for table in SOFT_DELETE_TABLES:
ensure_soft_delete_columns(table)
if "news" in tables:
for article in db["news"].find(status=None):
was_featured = article.get("featured", 0)
db["news"].update(
{
"uid": article["uid"],
"status": "published" if was_featured else "draft",
},
["uid"],
)
for article in db["news"].find(show_on_landing=None):
db["news"].update(
{
"uid": article["uid"],
"show_on_landing": 0,
},
["uid"],
)
for article in db["news"].find(slug=None):
from devplacepy.utils import make_combined_slug
slug = make_combined_slug(
article.get("title", "") or "news", article["uid"]
)
db["news"].update(
{
"uid": article["uid"],
"slug": slug,
},
["uid"],
)
if "news_sync" in tables:
for entry in db["news_sync"].find():
current = entry.get("status", "")
if current in ("below_threshold", ""):
db["news_sync"].update(
{
"id": entry["id"],
"status": "graded",
},
["id"],
)
if "site_settings" in tables:
news_defaults = {
"news_grade_threshold": "7",
"news_api_url": "https://news.app.molodetz.nl/api",
"news_ai_url": INTERNAL_GATEWAY_URL,
"news_ai_model": "molodetz",
}
for key, value in news_defaults.items():
existing = db["site_settings"].find_one(key=key)
if not existing:
db["site_settings"].insert(
{"uid": f"default_{key}", "key": key, "value": value}
)
upload_defaults = {
"max_upload_size_mb": "10",
"allowed_file_types": "",
"max_attachments_per_resource": "10",
}
for key, value in upload_defaults.items():
existing = db["site_settings"].find_one(key=key)
if not existing:
db["site_settings"].insert(
{"uid": f"default_{key}", "key": key, "value": value}
)
operational_defaults = {
"rate_limit_per_minute": "60",
"rate_limit_window_seconds": "60",
"news_service_interval": "3600",
"session_max_age_days": "7",
"session_remember_days": "30",
"registration_open": "1",
"maintenance_mode": "0",
"maintenance_message": "DevPlace is undergoing scheduled maintenance. Please check back shortly.",
"customization_enabled": "1",
"customization_js_enabled": "1",
"audit_log_retention_days": "90",
"docs_search_mode": "agent",
}
for key, value in operational_defaults.items():
existing = db["site_settings"].find_one(key=key)
if not existing:
db["site_settings"].insert(
{"uid": f"default_{key}", "key": key, "value": value}
)
_backfill_gamification()
backfill_api_keys()
migrate_ai_gateway_settings()
logger.info("Database initialized")
def internal_gateway_key() -> str:
return get_setting("gateway_internal_key", "")
OLD_GATEWAY_URL = "https://openai.app.molodetz.nl/v1/chat/completions"
def migrate_ai_gateway_settings() -> None:
import os
import uuid
if not get_setting("gateway_internal_key", ""):
set_setting("gateway_internal_key", str(uuid.uuid4()))
logger.info("Generated internal gateway key")
deepseek = os.environ.get("DEEPSEEK_API_KEY", "")
openrouter = os.environ.get("OPENROUTER_API_KEY", "")
if not get_setting("gateway_api_key", "") and (deepseek or openrouter):
set_setting("gateway_api_key", deepseek or openrouter)
logger.info("Migrated gateway upstream key from environment")
if not get_setting("gateway_vision_key", "") and openrouter:
set_setting("gateway_vision_key", openrouter)
logger.info("Migrated gateway vision key from environment")
for key in ("news_ai_url", "bot_api_url", "devii_ai_url"):
if get_setting(key, "") == OLD_GATEWAY_URL:
set_setting(key, INTERNAL_GATEWAY_URL)
logger.info(f"Migrated {key} to the internal gateway")
if get_setting("bot_model", "") == "deepseek-chat":
set_setting("bot_model", "molodetz")
logger.info("Migrated bot_model to molodetz")
def backfill_api_keys() -> int:
if "users" not in db.tables:
return 0
users = db["users"]
if not users.has_column("api_key"):
users.create_column_by_example("api_key", "")
if not users.has_column("cust_disable_global"):
users.create_column_by_example("cust_disable_global", 0)
if not users.has_column("cust_disable_pagetype"):
users.create_column_by_example("cust_disable_pagetype", 0)
import uuid_utils
updated = 0
for user in users.find():
if not user.get("api_key"):
users.update(
{"uid": user["uid"], "api_key": str(uuid_utils.uuid7())}, ["uid"]
)
updated += 1
return updated
def _backfill_gamification():
if "users" not in db.tables:
return
from devplacepy.utils import (
level_for_xp,
check_milestone_badges,
XP_POST,
XP_COMMENT,
XP_PROJECT,
XP_GIST,
XP_UPVOTE,
XP_FOLLOW,
)
pending = list(db["users"].find(xp=0))
if not pending:
return
xp_by_user = defaultdict(int)
def add_counts(table, column, points):
if table not in db.tables:
return
for row in db.query(
f"SELECT {column} AS uid, COUNT(*) AS c FROM {table} GROUP BY {column}"
):
if row["uid"]:
xp_by_user[row["uid"]] += row["c"] * points
add_counts("posts", "user_uid", XP_POST)
add_counts("comments", "user_uid", XP_COMMENT)
add_counts("projects", "user_uid", XP_PROJECT)
add_counts("gists", "user_uid", XP_GIST)
add_counts("follows", "following_uid", XP_FOLLOW)
if "votes" in db.tables:
for content_table, target_type in (
("posts", "post"),
("projects", "project"),
("gists", "gist"),
("comments", "comment"),
):
if content_table not in db.tables:
continue
rows = db.query(
f"SELECT c.user_uid AS uid, COUNT(*) AS c "
f"FROM votes v JOIN {content_table} c ON v.target_uid = c.uid "
f"WHERE v.target_type = :t AND v.value = 1 GROUP BY c.user_uid",
t=target_type,
)
for row in rows:
if row["uid"]:
xp_by_user[row["uid"]] += row["c"] * XP_UPVOTE
for user in pending:
xp = xp_by_user.get(user["uid"], 0)
if xp <= 0:
continue
db["users"].update(
{"uid": user["uid"], "xp": xp, "level": level_for_xp(xp)}, ["uid"]
)
_authors_cache.clear()
for user in pending:
check_milestone_badges(user["uid"])
logger.info(f"Gamification backfill processed {len(pending)} users")
def get_table(name):
return db[name]
def _in_clause(uids, prefix="p"):
placeholders = ", ".join(f":{prefix}{i}" for i in range(len(uids)))
params = {f"{prefix}{i}": uid for i, uid in enumerate(uids)}
return placeholders, params
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def soft_delete(table_name, deleted_by, *, stamp=None, **criteria):
if table_name not in db.tables:
return 0
table = db[table_name]
if not table.has_column("deleted_at"):
return 0
rows = list(table.find(deleted_at=None, **criteria))
if not rows:
return 0
stamp = stamp or _now_iso()
for row in rows:
table.update(
{"id": row["id"], "deleted_at": stamp, "deleted_by": deleted_by}, ["id"]
)
return len(rows)
def soft_delete_in(table_name, column, uids, deleted_by, *, stamp=None, **extra):
uids = [uid for uid in (uids or []) if uid]
if not uids or table_name not in db.tables:
return 0
if not db[table_name].has_column("deleted_at"):
return 0
placeholders, params = _in_clause(uids)
params["dat"] = stamp or _now_iso()
params["dby"] = deleted_by
extra_sql = ""
for index, (key, value) in enumerate(extra.items()):
params[f"x{index}"] = value
extra_sql += f" AND {key} = :x{index}"
db.query(
f"UPDATE {table_name} SET deleted_at = :dat, deleted_by = :dby "
f"WHERE {column} IN ({placeholders}) AND deleted_at IS NULL{extra_sql}",
**params,
)
return len(uids)
def restore(table_name, **criteria):
if table_name not in db.tables:
return 0
table = db[table_name]
if not table.has_column("deleted_at"):
return 0
rows = [row for row in table.find(**criteria) if row.get("deleted_at")]
for row in rows:
table.update({"id": row["id"], "deleted_at": None, "deleted_by": None}, ["id"])
return len(rows)
def purge(table_name, **criteria):
if table_name not in db.tables:
return 0
table = db[table_name]
count = table.count(**criteria)
table.delete(**criteria)
return int(count)
def list_deleted(table_name, page=1, per_page=25):
if table_name not in db.tables or not db[table_name].has_column("deleted_at"):
return [], build_pagination(page, 0, per_page)
table = db[table_name]
column = table.table.columns.deleted_at
total = table.count(column.isnot(None))
pagination = build_pagination(page, total, per_page)
offset = (pagination["page"] - 1) * pagination["per_page"]
rows = list(
table.find(
column.isnot(None),
order_by=["-deleted_at"],
_limit=pagination["per_page"],
_offset=offset,
)
)
return rows, pagination
def count_deleted(table_name):
if table_name not in db.tables or not db[table_name].has_column("deleted_at"):
return 0
table = db[table_name]
return int(table.count(table.table.columns.deleted_at.isnot(None)))
def restore_event(stamp):
if not stamp:
return 0
restored = 0
for table_name in SOFT_DELETE_TABLES:
if table_name in db.tables and db[table_name].has_column("deleted_at"):
restored += int(
db.query(
f"SELECT COUNT(*) AS n FROM {table_name} WHERE deleted_at = :s",
s=stamp,
).__next__()["n"]
)
db.query(
f"UPDATE {table_name} SET deleted_at = NULL, deleted_by = NULL "
f"WHERE deleted_at = :s",
s=stamp,
)
return restored
def purge_event(stamp):
if not stamp:
return []
purged = []
for table_name in SOFT_DELETE_TABLES:
if table_name in db.tables and db[table_name].has_column("deleted_at"):
rows = list(
db.query(
f"SELECT * FROM {table_name} WHERE deleted_at = :s", s=stamp
)
)
if rows:
purged.append((table_name, rows))
db.query(
f"DELETE FROM {table_name} WHERE deleted_at = :s", s=stamp
)
return purged
def get_users_by_uids(uids):
if not uids or "users" not in db.tables:
return {}
users = db["users"]
if "uid" not in users.columns:
return {}
seen = set()
unique = [u for u in uids if u not in seen and not seen.add(u)]
return {u["uid"]: u for u in users.find(users.table.columns.uid.in_(unique))}
def search_users_by_username(q, *, exclude_uid=None, limit=10):
if not q or "users" not in db.tables:
return []
if exclude_uid is not None:
rows = db.query(
"SELECT uid, username FROM users WHERE username LIKE :q AND uid != :me LIMIT :limit",
q=f"%{q}%",
me=exclude_uid,
limit=limit,
)
else:
rows = db.query(
"SELECT uid, username FROM users WHERE username LIKE :q LIMIT :limit",
q=f"%{q}%",
limit=limit,
)
return [{"uid": r["uid"], "username": r["username"]} for r in rows]
def get_comment_counts_by_post_uids(post_uids):
if not post_uids or "comments" not in db.tables:
return {}
placeholders, params = _in_clause(post_uids)
rows = db.query(
f"SELECT target_uid, COUNT(*) as c FROM comments WHERE target_type='post' AND target_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY target_uid",
**params,
)
return {r["target_uid"]: r["c"] for r in rows}
def get_post_counts_by_user_uids(user_uids):
if not user_uids or "posts" not in db.tables:
return {}
placeholders, params = _in_clause(user_uids)
rows = db.query(
f"SELECT user_uid, COUNT(*) as c FROM posts WHERE user_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY user_uid",
**params,
)
return {r["user_uid"]: r["c"] for r in rows}
def get_vote_counts(target_uids):
if not target_uids or "votes" not in db.tables:
return {}, {}
placeholders, params = _in_clause(target_uids)
rows = db.query(
f"SELECT target_uid, value, COUNT(*) as c FROM votes WHERE target_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY target_uid, value",
**params,
)
ups = {}
downs = {}
for r in rows:
if r["value"] == 1:
ups[r["target_uid"]] = r["c"]
else:
downs[r["target_uid"]] = r["c"]
return ups, downs
def get_user_votes(user_uid, target_uids):
if not user_uid or not target_uids or "votes" not in db.tables:
return {}
placeholders, params = _in_clause(target_uids)
params["uid"] = user_uid
rows = db.query(
f"SELECT target_uid, value FROM votes WHERE user_uid = :uid AND target_uid IN ({placeholders}) AND deleted_at IS NULL",
**params,
)
return {r["target_uid"]: r["value"] for r in rows}
def get_reactions_by_targets(target_type, target_uids, user=None):
if not target_uids or "reactions" not in db.tables:
return {}
placeholders, params = _in_clause(target_uids)
params["tt"] = target_type
rows = db.query(
f"SELECT target_uid, emoji, COUNT(*) as c FROM reactions WHERE target_type=:tt AND target_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY target_uid, emoji",
**params,
)
counts = defaultdict(dict)
for row in rows:
counts[row["target_uid"]][row["emoji"]] = row["c"]
mine = defaultdict(list)
if user:
placeholders, params = _in_clause(target_uids, prefix="m")
params["tt"] = target_type
params["u"] = user["uid"]
for row in db.query(
f"SELECT target_uid, emoji FROM reactions WHERE user_uid=:u AND target_type=:tt AND target_uid IN ({placeholders}) AND deleted_at IS NULL",
**params,
):
mine[row["target_uid"]].append(row["emoji"])
result = {}
for uid in target_uids:
result[uid] = {
"counts": dict(counts.get(uid, {})),
"mine": list(mine.get(uid, [])),
}
return result
def get_user_bookmarks(user_uid, target_type, target_uids):
if not user_uid or not target_uids or "bookmarks" not in db.tables:
return set()
placeholders, params = _in_clause(target_uids)
params["u"] = user_uid
params["tt"] = target_type
rows = db.query(
f"SELECT target_uid FROM bookmarks WHERE user_uid=:u AND target_type=:tt AND target_uid IN ({placeholders}) AND deleted_at IS NULL",
**params,
)
return {row["target_uid"] for row in rows}
def get_polls_by_post_uids(post_uids, user=None):
if not post_uids or "polls" not in db.tables:
return {}
placeholders, params = _in_clause(post_uids)
polls = list(
db.query(
f"SELECT * FROM polls WHERE post_uid IN ({placeholders}) AND deleted_at IS NULL",
**params,
)
)
if not polls:
return {}
poll_uids = [poll["uid"] for poll in polls]
option_placeholders, option_params = _in_clause(poll_uids, prefix="o")
options = list(
db.query(
f"SELECT * FROM poll_options WHERE poll_uid IN ({option_placeholders}) AND deleted_at IS NULL ORDER BY position",
**option_params,
)
)
counts = defaultdict(dict)
totals = defaultdict(int)
if "poll_votes" in db.tables:
vote_placeholders, vote_params = _in_clause(poll_uids, prefix="v")
for row in db.query(
f"SELECT poll_uid, option_uid, COUNT(*) as c FROM poll_votes WHERE poll_uid IN ({vote_placeholders}) AND deleted_at IS NULL GROUP BY poll_uid, option_uid",
**vote_params,
):
counts[row["poll_uid"]][row["option_uid"]] = row["c"]
totals[row["poll_uid"]] += row["c"]
user_choice = {}
if user and "poll_votes" in db.tables:
placeholders, params = _in_clause(poll_uids, prefix="m")
params["u"] = user["uid"]
for row in db.query(
f"SELECT poll_uid, option_uid FROM poll_votes WHERE user_uid=:u AND poll_uid IN ({placeholders}) AND deleted_at IS NULL",
**params,
):
user_choice[row["poll_uid"]] = row["option_uid"]
options_by_poll = defaultdict(list)
for option in options:
options_by_poll[option["poll_uid"]].append(option)
result = {}
for poll in polls:
poll_uid = poll["uid"]
total = totals.get(poll_uid, 0)
rendered = []
for option in options_by_poll.get(poll_uid, []):
count = counts.get(poll_uid, {}).get(option["uid"], 0)
rendered.append(
{
"uid": option["uid"],
"label": option["label"],
"count": count,
"pct": round(count * 100 / total) if total else 0,
}
)
result[poll["post_uid"]] = {
"uid": poll_uid,
"question": poll["question"],
"options": rendered,
"total": total,
"my_choice": user_choice.get(poll_uid),
}
return result
def get_poll_for_post(post_uid, user=None):
return get_polls_by_post_uids([post_uid], user).get(post_uid)
def _build_comment_items(raw, user=None):
uids = [c["user_uid"] for c in raw]
cids = [c["uid"] for c in raw]
users = get_users_by_uids(uids)
ups, downs = get_vote_counts(cids)
user_votes = get_user_votes(user["uid"], cids) if user else {}
reactions = get_reactions_by_targets("comment", cids, user)
from devplacepy.utils import time_ago
from devplacepy.attachments import get_attachments_batch as _gab
atts_map = _gab("comment", cids) if "attachments" in db.tables else {}
items = {}
for c in raw:
items[c["uid"]] = {
"comment": c,
"author": users.get(c["user_uid"]),
"time_ago": time_ago(c["created_at"]),
"votes": {"up": ups.get(c["uid"], 0), "down": downs.get(c["uid"], 0)},
"my_vote": user_votes.get(c["uid"], 0),
"children": [],
"attachments": atts_map.get(c["uid"], []),
"reactions": reactions.get(c["uid"], {"counts": {}, "mine": []}),
}
return items
def load_comments(target_type, target_uid, user=None):
if "comments" not in db.tables:
return []
comments_table = db["comments"]
raw = list(
comments_table.find(
target_type=target_type,
target_uid=target_uid,
deleted_at=None,
order_by=["created_at"],
)
)
if not raw and target_type == "post":
raw = list(
comments_table.find(
post_uid=target_uid, deleted_at=None, order_by=["created_at"]
)
)
if not raw:
return []
cmap = _build_comment_items(raw, user)
top = []
for item in cmap.values():
parent = item["comment"].get("parent_uid")
if parent and parent in cmap:
cmap[parent]["children"].append(item)
else:
top.append(item)
return top
def get_recent_comments_by_post_uids(post_uids, limit=3, user=None):
if not post_uids or "comments" not in db.tables:
return {}
placeholders, params = _in_clause(post_uids)
params["lim"] = limit
raw = list(
db.query(
f"SELECT * FROM ("
f" SELECT *, ROW_NUMBER() OVER ("
f" PARTITION BY target_uid ORDER BY created_at DESC, id DESC"
f" ) AS rn FROM comments"
f" WHERE target_type='post' AND target_uid IN ({placeholders}) AND deleted_at IS NULL"
f") WHERE rn <= :lim ORDER BY target_uid, created_at ASC",
**params,
)
)
if not raw:
return {}
items = _build_comment_items(raw, user)
result = defaultdict(list)
for c in raw:
result[c["target_uid"]].append(items[c["uid"]])
return dict(result)
def load_comments_by_target_uids(target_type, target_uids, user=None):
if not target_uids or "comments" not in db.tables:
return {}
placeholders, params = _in_clause(target_uids)
params["tt"] = target_type
raw = list(
db.query(
f"SELECT * FROM comments WHERE target_type=:tt AND target_uid IN ({placeholders}) ORDER BY created_at",
**params,
)
)
if not raw:
return {}
from collections import defaultdict
by_uid = defaultdict(list)
for c in raw:
by_uid[c["target_uid"]].append(c)
result = {}
for uid in target_uids:
group = by_uid.get(uid, [])
if not group:
result[uid] = []
continue
cmap = _build_comment_items(group, user)
tree = []
for item in cmap.values():
parent = item["comment"].get("parent_uid")
if parent and parent in cmap:
cmap[parent]["children"].append(item)
else:
tree.append(item)
result[uid] = tree
return result
def get_attachments(resource_type: str, resource_uid: str) -> list:
if "attachments" not in db.tables:
return []
return list(
db["attachments"].find(
resource_type=resource_type,
resource_uid=resource_uid,
deleted_at=None,
order_by=["created_at"],
)
)
def get_attachments_by_type(resource_type: str, resource_uids: list) -> dict:
if not resource_uids or "attachments" not in db.tables:
return {}
rows = list(
db["attachments"].find(
db["attachments"].table.columns.resource_uid.in_(resource_uids),
db["attachments"].table.columns.deleted_at.is_(None),
resource_type=resource_type,
)
)
result = {}
for a in rows:
key = a["resource_uid"]
if key not in result:
result[key] = []
result[key].append(a)
return result
def get_news_images_by_uids(news_uids: list) -> dict:
if not news_uids or "news_images" not in db.tables:
return {}
images_table = db["news_images"]
if not images_table.has_column("news_uid"):
return {}
rows = images_table.find(
images_table.table.columns.news_uid.in_(news_uids),
images_table.table.columns.deleted_at.is_(None),
order_by=["uid"],
)
result = {}
for r in rows:
result.setdefault(r["news_uid"], r["url"])
return result
def delete_attachment_record(uid: str) -> None:
if "attachments" not in db.tables:
return
att = db["attachments"].find_one(uid=uid)
if att:
_delete_attachment_file(att.get("storage_path", ""))
db["attachments"].delete(id=att["id"])
def delete_attachments(resource_type: str, resource_uid: str) -> None:
if "attachments" not in db.tables:
return
for a in db["attachments"].find(
resource_type=resource_type, resource_uid=resource_uid
):
_delete_attachment_file(a.get("storage_path", ""))
db["attachments"].delete(resource_type=resource_type, resource_uid=resource_uid)
def _delete_attachment_file(storage_path: str) -> None:
if not storage_path:
return
from devplacepy.config import UPLOADS_DIR
file_path = UPLOADS_DIR / storage_path
try:
file_path.unlink(missing_ok=True)
parent = file_path.parent
if parent.exists() and not any(parent.iterdir()):
parent.rmdir()
grandparent = parent.parent
if grandparent.exists() and not any(grandparent.iterdir()):
grandparent.rmdir()
except Exception as e:
logger.warning(f"Failed to delete attachment file {storage_path}: {e}")
_settings_cache = TTLCache(ttl=60, max_size=100)
def get_setting(key: str, default: str = "") -> str:
sync_local_cache("settings", _settings_cache)
cached = _settings_cache.get(key)
if cached is not None:
return cached
if "site_settings" not in db.tables:
return default
entry = db["site_settings"].find_one(key=key)
if entry is None:
return default
_settings_cache.set(key, entry["value"])
return entry["value"]
def get_int_setting(key: str, default: int) -> int:
raw = get_setting(key, str(default))
try:
return int(raw)
except (TypeError, ValueError):
return default
def set_setting(key: str, value: str) -> None:
settings = get_table("site_settings")
existing = settings.find_one(key=key)
if existing:
settings.update({"id": existing["id"], "key": key, "value": value}, ["id"])
else:
settings.insert({"uid": f"setting_{key}", "key": key, "value": value})
_settings_cache.set(key, value)
bump_cache_version("settings")
def clear_settings_cache() -> None:
_settings_cache.clear()
bump_cache_version("settings")
CUSTOMIZATION_GLOBAL_SCOPE = "global"
CUSTOMIZATION_LANGS = ("css", "js")
_customizations_cache = TTLCache(ttl=300, max_size=100)
def _customization_key(owner_kind: str, owner_id: str, page_type: str) -> str:
return f"{owner_kind}\x1f{owner_id}\x1f{page_type}"
CUSTOMIZATION_PREF_COLUMNS = {
"global": "cust_disable_global",
"pagetype": "cust_disable_pagetype",
}
def get_customization_prefs(owner_kind: str, owner_id: str) -> dict:
if owner_kind != "user" or "users" not in db.tables:
return {"disable_global": False, "disable_pagetype": False}
user = db["users"].find_one(uid=owner_id)
if user is None:
return {"disable_global": False, "disable_pagetype": False}
return {
"disable_global": bool(user.get("cust_disable_global", 0)),
"disable_pagetype": bool(user.get("cust_disable_pagetype", 0)),
}
def set_customization_pref(owner_id: str, category: str, disabled: bool) -> None:
column = CUSTOMIZATION_PREF_COLUMNS.get(category)
if column is None:
raise ValueError(f"Unknown customization category: {category}")
from devplacepy.utils import clear_user_cache
get_table("users").update(
{"uid": owner_id, column: 1 if disabled else 0}, ["uid"]
)
clear_user_cache(owner_id)
bump_cache_version("customizations")
def get_custom_overrides(owner_kind: str, owner_id: str, page_type: str) -> dict:
sync_local_cache("customizations", _customizations_cache)
key = _customization_key(owner_kind, owner_id, page_type)
cached = _customizations_cache.get(key)
if cached is not None:
return cached
result = {"css": "", "js": ""}
if "user_customizations" in db.tables:
prefs = get_customization_prefs(owner_kind, owner_id)
scopes = (CUSTOMIZATION_GLOBAL_SCOPE, page_type)
rows = db["user_customizations"].find(
owner_kind=owner_kind,
owner_id=owner_id,
enabled=1,
deleted_at=None,
)
pieces: dict[str, dict[str, str]] = {lang: {} for lang in CUSTOMIZATION_LANGS}
for row in rows:
lang = row.get("lang")
scope = row.get("scope")
if lang not in pieces or scope not in scopes:
continue
if scope == CUSTOMIZATION_GLOBAL_SCOPE and prefs["disable_global"]:
continue
if scope != CUSTOMIZATION_GLOBAL_SCOPE and prefs["disable_pagetype"]:
continue
pieces[lang][scope] = row.get("code") or ""
for lang in CUSTOMIZATION_LANGS:
ordered = [pieces[lang][scope] for scope in scopes if scope in pieces[lang]]
result[lang] = "\n".join(part for part in ordered if part.strip())
_customizations_cache.set(key, result)
return result
def get_custom_override(
owner_kind: str, owner_id: str, scope: str, lang: str
) -> dict | None:
if "user_customizations" not in db.tables:
return None
return db["user_customizations"].find_one(
owner_kind=owner_kind,
owner_id=owner_id,
scope=scope,
lang=lang,
deleted_at=None,
)
def list_custom_overrides(owner_kind: str, owner_id: str) -> list:
if "user_customizations" not in db.tables:
return []
return list(
db["user_customizations"].find(
owner_kind=owner_kind, owner_id=owner_id, deleted_at=None
)
)
def set_custom_override(
owner_kind: str, owner_id: str, scope: str, lang: str, code: str
) -> dict:
from devplacepy.utils import generate_uid
table = get_table("user_customizations")
now = datetime.now(timezone.utc).isoformat()
existing = table.find_one(
owner_kind=owner_kind, owner_id=owner_id, scope=scope, lang=lang
)
if existing:
record = {
"id": existing["id"],
"code": code,
"enabled": 1,
"updated_at": now,
"deleted_at": None,
"deleted_by": None,
}
table.update(record, ["id"])
result = {**existing, **record}
else:
result = {
"uid": generate_uid(),
"owner_kind": owner_kind,
"owner_id": owner_id,
"scope": scope,
"lang": lang,
"code": code,
"enabled": 1,
"created_at": now,
"updated_at": now,
"deleted_at": None,
"deleted_by": None,
}
table.insert(result)
bump_cache_version("customizations")
return result
def delete_custom_override(
owner_kind: str,
owner_id: str,
scope: str | None = None,
lang: str | None = None,
deleted_by: str | None = None,
) -> int:
if "user_customizations" not in db.tables:
return 0
criteria: dict = {"owner_kind": owner_kind, "owner_id": owner_id}
if scope is not None:
criteria["scope"] = scope
if lang is not None:
criteria["lang"] = lang
count = soft_delete(
"user_customizations", deleted_by or f"{owner_kind}:{owner_id}", **criteria
)
bump_cache_version("customizations")
return int(count)
NOTIFICATION_TYPES = [
{"key": "comment", "label": "Comments", "description": "Someone comments on your post"},
{"key": "reply", "label": "Replies", "description": "Someone replies to your comment"},
{"key": "mention", "label": "Mentions", "description": "Someone mentions you with @username"},
{"key": "vote", "label": "Upvotes", "description": "Someone ++'d your content"},
{"key": "follow", "label": "Followers", "description": "Someone starts following you"},
{"key": "message", "label": "Direct messages", "description": "Someone sends you a message"},
{"key": "badge", "label": "Badges", "description": "You earn a badge"},
{"key": "level", "label": "Level-ups", "description": "You reach a new level"},
{"key": "issue", "label": "Issue tracker", "description": "Updates on issue reports you filed"},
]
NOTIFICATION_CHANNELS = ("in_app", "push")
_NOTIFICATION_CHANNEL_COLUMNS = {"in_app": "in_app_enabled", "push": "push_enabled"}
_NOTIFICATION_TYPE_KEYS = {entry["key"] for entry in NOTIFICATION_TYPES}
_notification_prefs_cache = TTLCache(ttl=300, max_size=500)
def _notification_default(notification_type: str, channel: str) -> bool:
return get_int_setting(f"notif_default_{notification_type}_{channel}", 1) != 0
def get_notification_default(notification_type: str, channel: str) -> bool:
return _notification_default(notification_type, channel)
def set_notification_default(
notification_type: str, channel: str, enabled: bool
) -> None:
if channel not in _NOTIFICATION_CHANNEL_COLUMNS:
raise ValueError(f"Unknown notification channel: {channel}")
set_setting(f"notif_default_{notification_type}_{channel}", "1" if enabled else "0")
def _notification_overrides(user_uid: str) -> dict:
sync_local_cache("notif_prefs", _notification_prefs_cache)
cached = _notification_prefs_cache.get(user_uid)
if cached is not None:
return cached
overrides: dict = {}
if "notification_preferences" in db.tables:
for row in db["notification_preferences"].find(
user_uid=user_uid, deleted_at=None
):
overrides[row["notification_type"]] = {
"in_app": bool(row.get("in_app_enabled", 1)),
"push": bool(row.get("push_enabled", 1)),
}
_notification_prefs_cache.set(user_uid, overrides)
return overrides
def notification_enabled(user_uid: str, notification_type: str, channel: str) -> bool:
if channel not in _NOTIFICATION_CHANNEL_COLUMNS:
return True
override = _notification_overrides(user_uid).get(notification_type)
if override is not None:
return bool(override[channel])
return _notification_default(notification_type, channel)
def get_notification_prefs(user_uid: str) -> list:
overrides = _notification_overrides(user_uid)
result = []
for entry in NOTIFICATION_TYPES:
key = entry["key"]
override = overrides.get(key)
in_app = (
bool(override["in_app"]) if override else _notification_default(key, "in_app")
)
push = bool(override["push"]) if override else _notification_default(key, "push")
result.append(
{
"key": key,
"label": entry["label"],
"description": entry["description"],
"in_app": in_app,
"push": push,
"customized": override is not None,
}
)
return result
def set_notification_pref(
user_uid: str, notification_type: str, channel: str, enabled: bool
) -> dict:
if notification_type not in _NOTIFICATION_TYPE_KEYS:
raise ValueError(f"Unknown notification type: {notification_type}")
if channel not in _NOTIFICATION_CHANNEL_COLUMNS:
raise ValueError(f"Unknown notification channel: {channel}")
from devplacepy.utils import generate_uid
table = get_table("notification_preferences")
now = datetime.now(timezone.utc).isoformat()
existing = table.find_one(user_uid=user_uid, notification_type=notification_type)
if existing:
values = {
"in_app": bool(existing.get("in_app_enabled", 1)),
"push": bool(existing.get("push_enabled", 1)),
}
else:
values = {
"in_app": _notification_default(notification_type, "in_app"),
"push": _notification_default(notification_type, "push"),
}
values[channel] = enabled
if existing:
record = {
"id": existing["id"],
"in_app_enabled": 1 if values["in_app"] else 0,
"push_enabled": 1 if values["push"] else 0,
"updated_at": now,
"deleted_at": None,
"deleted_by": None,
}
table.update(record, ["id"])
result = {**existing, **record}
else:
result = {
"uid": generate_uid(),
"user_uid": user_uid,
"notification_type": notification_type,
"in_app_enabled": 1 if values["in_app"] else 0,
"push_enabled": 1 if values["push"] else 0,
"created_at": now,
"updated_at": now,
"deleted_at": None,
"deleted_by": None,
}
table.insert(result)
bump_cache_version("notif_prefs")
return result
def reset_notification_prefs(user_uid: str, deleted_by: str | None = None) -> int:
if "notification_preferences" not in db.tables:
return 0
count = soft_delete(
"notification_preferences", deleted_by or f"user:{user_uid}", user_uid=user_uid
)
bump_cache_version("notif_prefs")
return int(count)
_stats_cache = TTLCache(ttl=30, max_size=50)
def get_site_stats() -> dict:
cached = _stats_cache.get("site")
if cached is not None:
return cached
today_start = (
datetime.now(timezone.utc)
.replace(hour=0, minute=0, second=0, microsecond=0)
.isoformat()
)
stats = {
"total_members": db["users"].count() if "users" in db.tables else 0,
"posts_today": db["posts"].count(
created_at={">=": today_start}, deleted_at=None
)
if "posts" in db.tables
else 0,
"total_projects": db["projects"].count(deleted_at=None)
if "projects" in db.tables
else 0,
"total_gists": db["gists"].count(deleted_at=None)
if "gists" in db.tables
else 0,
}
_stats_cache.set("site", stats)
return stats
_analytics_cache = TTLCache(ttl=300, max_size=50)
def get_platform_analytics(top_n: int = 10) -> dict:
top_n = max(1, min(int(top_n or 10), 50))
cache_key = f"analytics:{top_n}"
cached = _analytics_cache.get(cache_key)
if cached is not None:
return cached
now = datetime.now(timezone.utc)
d1 = (now - timedelta(days=1)).isoformat()
d7 = (now - timedelta(days=7)).isoformat()
d30 = (now - timedelta(days=30)).isoformat()
active_24h = active_7d = active_30d = 0
sources = [
table
for table in ("posts", "comments", "gists", "projects")
if table in db.tables
]
if sources:
union = " UNION ALL ".join(
f"SELECT user_uid, created_at FROM {table} WHERE deleted_at IS NULL"
for table in sources
)
rows = db.query(
"SELECT "
"COUNT(DISTINCT CASE WHEN created_at >= :d1 THEN user_uid END) AS a1, "
"COUNT(DISTINCT CASE WHEN created_at >= :d7 THEN user_uid END) AS a7, "
"COUNT(DISTINCT CASE WHEN created_at >= :d30 THEN user_uid END) AS a30 "
f"FROM ({union})",
d1=d1,
d7=d7,
d30=d30,
)
for row in rows:
active_24h = row["a1"] or 0
active_7d = row["a7"] or 0
active_30d = row["a30"] or 0
signed_in_now = 0
if "sessions" in db.tables:
for row in db.query(
"SELECT COUNT(DISTINCT user_uid) AS n FROM sessions WHERE expires_at > :now AND deleted_at IS NULL",
now=now.isoformat(),
):
signed_in_now = row["n"] or 0
def _users_created_since(cutoff: str) -> int:
return (
db["users"].count(created_at={">=": cutoff}) if "users" in db.tables else 0
)
site = get_site_stats()
totals = {
"posts": db["posts"].count(deleted_at=None) if "posts" in db.tables else 0,
"comments": db["comments"].count(deleted_at=None)
if "comments" in db.tables
else 0,
"gists": site["total_gists"],
"projects": site["total_projects"],
"news": db["news"].count(deleted_at=None) if "news" in db.tables else 0,
}
top_authors = [
{"username": author.get("username", ""), "stars": author.get("stars", 0)}
for author in get_top_authors(top_n)
]
result = {
"total_members": site["total_members"],
"active_24h": active_24h,
"active_7d": active_7d,
"active_30d": active_30d,
"signed_in_now": signed_in_now,
"new_24h": _users_created_since(d1),
"new_7d": _users_created_since(d7),
"new_30d": _users_created_since(d30),
"posts_today": site["posts_today"],
"totals": totals,
"top_authors": top_authors,
"active_definition": (
"Active = created a post, comment, gist, or project within the window. "
"Automated/bot accounts are not separately flagged."
),
"signed_in_definition": (
"signed_in_now = distinct members holding an unexpired session cookie, i.e. who "
"logged in within the session lifetime (default 7 days, up to 30 with remember-me) "
"and have not logged out. It is NOT a real-time presence/online count - the platform "
"does not track per-request last-seen activity - so it is normally a large fraction "
"of recently active members. Do not describe it as 'currently online' or 'logged in "
"right now'."
),
}
_analytics_cache.set(cache_key, result)
return result
_activity_cache = TTLCache(ttl=300, max_size=1000)
_ACTIVITY_TABLES = ("posts", "comments", "gists", "projects")
def get_activity_calendar(user_uid: str) -> dict:
cached = _activity_cache.get(user_uid)
if cached is not None:
return cached
sources = [table for table in _ACTIVITY_TABLES if table in db.tables]
calendar: dict[str, int] = {}
if sources:
cutoff = (datetime.now(timezone.utc) - timedelta(days=364)).date().isoformat()
union = " UNION ALL ".join(
f"SELECT created_at FROM {table} WHERE user_uid = :u AND deleted_at IS NULL"
for table in sources
)
rows = db.query(
f"SELECT date(created_at) AS day, COUNT(*) AS c FROM ({union}) WHERE date(created_at) >= :cutoff GROUP BY day",
u=user_uid,
cutoff=cutoff,
)
for row in rows:
if row["day"]:
calendar[row["day"]] = row["c"]
_activity_cache.set(user_uid, calendar)
return calendar
def _activity_level(count: int) -> int:
if count <= 0:
return 0
if count == 1:
return 1
if count <= 3:
return 2
if count <= 6:
return 3
return 4
def get_first_activity_date(user_uid: str):
sources = [table for table in _ACTIVITY_TABLES if table in db.tables]
if not sources:
return None
union = " UNION ALL ".join(
f"SELECT MIN(created_at) AS m FROM {table} WHERE user_uid = :u AND deleted_at IS NULL"
for table in sources
)
for row in db.query(f"SELECT MIN(m) AS first FROM ({union})", u=user_uid):
if row["first"]:
return datetime.fromisoformat(row["first"]).date()
return None
HEATMAP_WEEKS = 53
def get_activity_heatmap(user_uid: str) -> list:
calendar = get_activity_calendar(user_uid)
today = datetime.now(timezone.utc).date()
week_start = today - timedelta(days=today.weekday())
start = week_start - timedelta(weeks=HEATMAP_WEEKS - 1)
first = get_first_activity_date(user_uid)
if first:
first_week = first - timedelta(days=first.weekday())
if first_week > start:
start = first_week
weeks = []
for w in range(HEATMAP_WEEKS):
week = []
for d in range(7):
day = start + timedelta(days=w * 7 + d)
iso = day.isoformat()
count = calendar.get(iso, 0)
week.append({"date": iso, "count": count, "level": _activity_level(count)})
weeks.append(week)
return weeks
def get_activity_months(weeks: list) -> list:
if not weeks:
return []
last = len(weeks) - 1
labels = []
for i in range(6):
column = round(i * last / 5)
iso = weeks[column][0]["date"]
labels.append(datetime.fromisoformat(iso).strftime("%b"))
return labels
def get_streaks(user_uid: str) -> dict:
calendar = get_activity_calendar(user_uid)
if not calendar:
return {"current": 0, "longest": 0}
dates = sorted(datetime.fromisoformat(day).date() for day in calendar)
date_set = set(dates)
longest = 1
run = 1
for index in range(1, len(dates)):
if (dates[index] - dates[index - 1]).days == 1:
run += 1
else:
run = 1
longest = max(longest, run)
today = datetime.now(timezone.utc).date()
cursor = today
if today not in date_set and (today - timedelta(days=1)) in date_set:
cursor = today - timedelta(days=1)
current = 0
while cursor in date_set:
current += 1
cursor = cursor - timedelta(days=1)
return {"current": current, "longest": longest}
_gist_languages_cache = TTLCache(ttl=60, max_size=200)
def get_gist_languages() -> set[str]:
cached = _gist_languages_cache.get("codes")
if cached is not None:
return cached
codes: set[str] = set()
if "gists" in db.tables:
for row in db.query(
"SELECT DISTINCT language FROM gists WHERE deleted_at IS NULL"
):
language = row.get("language")
if language:
codes.add(language)
_gist_languages_cache.set("codes", codes)
return codes
VOTABLE_TARGETS: dict[str, str] = {
"post": "posts",
"project": "projects",
"gist": "gists",
"comment": "comments",
}
STAR_TARGETS: set[str] = {"post", "project", "gist"}
_authors_cache = TTLCache(ttl=60, max_size=200)
def _ranked_authors() -> list:
cached = _authors_cache.get("ranked")
if cached is not None:
return cached
sources = [
(target_type, table_name)
for target_type, table_name in VOTABLE_TARGETS.items()
if table_name in db.tables
]
if "votes" not in db.tables or not sources:
_authors_cache.set("ranked", [])
return []
target_union = " UNION ALL ".join(
f"SELECT uid, user_uid, '{target_type}' AS target_type FROM {table_name} WHERE deleted_at IS NULL"
for target_type, table_name in sources
)
rows = db.query(
f"SELECT t.user_uid, SUM(v.value) AS total "
f"FROM votes v JOIN ({target_union}) t ON v.target_uid = t.uid AND v.target_type = t.target_type "
f"WHERE v.deleted_at IS NULL "
f"GROUP BY t.user_uid HAVING SUM(v.value) > 0 ORDER BY total DESC"
)
ranked = [(row["user_uid"], row["total"]) for row in rows]
users_map = get_users_by_uids([uid for uid, _ in ranked])
authors = []
for uid, total in ranked:
user = users_map.get(uid)
if user:
author = dict(user)
author["stars"] = total
authors.append(author)
_authors_cache.set("ranked", authors)
return authors
def get_top_authors(limit: int = 5) -> list:
return _ranked_authors()[:limit]
def get_leaderboard(limit: int = 50, offset: int = 0) -> list:
sliced = _ranked_authors()[offset : offset + limit]
leaderboard = []
for position, author in enumerate(sliced, start=offset + 1):
entry = dict(author)
entry["rank"] = position
leaderboard.append(entry)
return leaderboard
def get_user_rank(user_uid: str):
for position, author in enumerate(_ranked_authors(), start=1):
if author["uid"] == user_uid:
return position
return None
def get_user_stars(user_uid: str) -> int:
if "votes" not in db.tables:
return 0
target_union = " UNION ALL ".join(
f"SELECT uid, '{target_type}' AS target_type FROM {table_name} WHERE user_uid = :u AND deleted_at IS NULL"
for target_type, table_name in VOTABLE_TARGETS.items()
if table_name in db.tables
)
if not target_union:
return 0
for row in db.query(
f"SELECT COALESCE(SUM(v.value), 0) AS s "
f"FROM votes v JOIN ({target_union}) t ON v.target_uid = t.uid AND v.target_type = t.target_type "
f"WHERE v.deleted_at IS NULL",
u=user_uid,
):
return row["s"] or 0
return 0
def resolve_by_slug(table, slug, include_deleted=False):
has_soft_delete = table.has_column("deleted_at")
flt = {} if include_deleted or not has_soft_delete else {"deleted_at": None}
entry = table.find_one(slug=slug, **flt)
if not entry:
entry = table.find_one(uid=slug, **flt)
return entry
def record_fork(
source_project_uid: str, forked_project_uid: str, forked_by_uid: str
) -> None:
from devplacepy.utils import generate_uid
get_table("project_forks").insert(
{
"uid": generate_uid(),
"source_project_uid": source_project_uid,
"forked_project_uid": forked_project_uid,
"forked_by_uid": forked_by_uid,
"created_at": datetime.now(timezone.utc).isoformat(),
"deleted_at": None,
"deleted_by": None,
}
)
def get_fork_parent(forked_project_uid: str) -> dict | None:
if "project_forks" not in db.tables:
return None
relation = get_table("project_forks").find_one(
forked_project_uid=forked_project_uid, deleted_at=None
)
if not relation:
return None
return get_table("projects").find_one(
uid=relation["source_project_uid"], deleted_at=None
)
def count_forks(source_project_uid: str) -> int:
if "project_forks" not in db.tables:
return 0
return get_table("project_forks").count(
source_project_uid=source_project_uid, deleted_at=None
)
def soft_delete_fork_relations(project_uid: str, deleted_by: str) -> None:
if "project_forks" not in db.tables:
return
stamp = _now_iso()
soft_delete("project_forks", deleted_by, stamp=stamp, forked_project_uid=project_uid)
soft_delete("project_forks", deleted_by, stamp=stamp, source_project_uid=project_uid)
def delete_fork_relations(project_uid: str) -> None:
if "project_forks" not in db.tables:
return
forks = get_table("project_forks")
forks.delete(forked_project_uid=project_uid)
forks.delete(source_project_uid=project_uid)
def resolve_object_url(target_type: str, target_uid: str) -> str:
if target_type == "post":
post = resolve_by_slug(get_table("posts"), target_uid)
return f"/posts/{post['slug'] or post['uid']}" if post else "/feed"
if target_type == "project":
project = resolve_by_slug(get_table("projects"), target_uid)
return (
f"/projects/{project['slug'] or project['uid']}" if project else "/projects"
)
if target_type == "news":
article = resolve_by_slug(get_table("news"), target_uid)
if article:
return f"/news/{article.get('slug', '') or article['uid']}"
return "/news"
if target_type == "issue":
return f"/issues?highlight={target_uid}"
if target_type == "gist":
gist = resolve_by_slug(get_table("gists"), target_uid)
return f"/gists/{gist['slug'] or gist['uid']}" if gist else "/gists"
if target_type == "comment":
comment = get_table("comments").find_one(uid=target_uid, deleted_at=None)
if not comment:
return "/feed"
parent_url = resolve_object_url(
comment.get("target_type", "post"),
comment.get("target_uid") or comment.get("post_uid", ""),
)
return f"{parent_url}#comment-{target_uid}"
return "/feed"
def update_target_stars(target_type: str, target_uid: str, net_stars: int) -> None:
table_name = VOTABLE_TARGETS.get(target_type)
if not table_name:
return
if target_type in STAR_TARGETS:
get_table(table_name).update({"uid": target_uid, "stars": net_stars}, ["uid"])
_authors_cache.clear()
def soft_delete_engagement(target_type: str, target_uids: list, deleted_by: str) -> None:
uids = [uid for uid in (target_uids or []) if uid]
if not uids:
return
stamp = _now_iso()
soft_delete_in(
"reactions", "target_uid", uids, deleted_by, stamp=stamp, target_type=target_type
)
soft_delete_in(
"bookmarks", "target_uid", uids, deleted_by, stamp=stamp, target_type=target_type
)
if target_type == "post" and "polls" in db.tables:
for uid in uids:
for poll in db["polls"].find(post_uid=uid, deleted_at=None):
soft_delete("poll_votes", deleted_by, stamp=stamp, poll_uid=poll["uid"])
soft_delete("poll_options", deleted_by, stamp=stamp, poll_uid=poll["uid"])
soft_delete("polls", deleted_by, stamp=stamp, post_uid=uid)
def delete_engagement(target_type: str, target_uids: list) -> None:
uids = [uid for uid in (target_uids or []) if uid]
if not uids:
return
if "reactions" in db.tables:
placeholders, params = _in_clause(uids)
params["tt"] = target_type
db.query(
f"DELETE FROM reactions WHERE target_type=:tt AND target_uid IN ({placeholders})",
**params,
)
if "bookmarks" in db.tables:
placeholders, params = _in_clause(uids)
params["tt"] = target_type
db.query(
f"DELETE FROM bookmarks WHERE target_type=:tt AND target_uid IN ({placeholders})",
**params,
)
if target_type == "post" and "polls" in db.tables:
for uid in uids:
for poll in db["polls"].find(post_uid=uid):
if "poll_votes" in db.tables:
db["poll_votes"].delete(poll_uid=poll["uid"])
if "poll_options" in db.tables:
db["poll_options"].delete(poll_uid=poll["uid"])
db["polls"].delete(post_uid=uid)
def get_target_owner_uid(target_type: str, target_uid: str) -> str | None:
table_name = VOTABLE_TARGETS.get(target_type)
if not table_name:
return None
row = get_table(table_name).find_one(uid=target_uid, deleted_at=None)
return row["user_uid"] if row else None
PAGE_SIZE = 25
def text_search_clause(table, search, fields=("title", "description")):
if not search or not search.strip() or not table.exists:
return None
columns = table.table.columns
like = f"%{search.strip()}%"
matches = [columns[field].ilike(like) for field in fields if field in columns]
return or_(*matches) if matches else None
def paginate(
table, *clauses, before=None, order=None, cursor_field="created_at", **filters
):
order = order or ["-" + cursor_field]
clauses = list(clauses)
if table.has_column("deleted_at") and "deleted_at" not in filters:
clauses.append(table.table.columns.deleted_at.is_(None))
if before:
clauses.append(table.table.columns[cursor_field] < before)
rows = list(table.find(*clauses, **filters, order_by=order, _limit=PAGE_SIZE + 1))
has_more = len(rows) > PAGE_SIZE
rows = rows[:PAGE_SIZE]
next_cursor = rows[-1][cursor_field] if has_more and rows else None
return rows, next_cursor
def interleave_by_author(rows, uid_key="user_uid"):
queues: dict = {}
appearance: list = []
for position, row in enumerate(rows):
owner = row.get(uid_key)
queue = queues.get(owner)
if queue is None:
queue = queues[owner] = []
appearance.append(owner)
queue.append((position, row))
spread = []
last_owner = object()
while len(spread) < len(rows):
chosen = None
for owner in appearance:
queue = queues[owner]
if not queue or owner == last_owner:
continue
count = len(queue)
head_position = queue[0][0]
if (
chosen is None
or count > chosen[0]
or (count == chosen[0] and head_position < chosen[1])
):
chosen = (count, head_position, owner)
owner = last_owner if chosen is None else chosen[2]
spread.append(queues[owner].pop(0)[1])
last_owner = owner
return spread
def paginate_diverse(
table,
*clauses,
before=None,
order=None,
cursor_field="created_at",
uid_key="user_uid",
**filters,
):
rows, next_cursor = paginate(
table,
*clauses,
before=before,
order=order,
cursor_field=cursor_field,
**filters,
)
return interleave_by_author(rows, uid_key=uid_key), next_cursor
def get_user_post_count(user_uid: str) -> int:
if "posts" not in db.tables:
return 0
return get_table("posts").count(user_uid=user_uid, deleted_at=None)
def build_pagination(page, total, per_page=25):
total_pages = max(1, __import__("math").ceil(total / per_page))
page = max(1, min(page, total_pages))
return {
"page": page,
"per_page": per_page,
"total": total,
"total_pages": total_pages,
"has_prev": page > 1,
"has_next": page < total_pages,
"prev_page": page - 1,
"next_page": page + 1,
}
def get_user_media(user_uid: str, page: int = 1, per_page: int = 24) -> tuple:
if "attachments" not in db.tables:
return [], build_pagination(page, 0, per_page)
from devplacepy.attachments import _row_to_attachment
total = list(
db.query(
"SELECT COUNT(*) AS n FROM attachments "
"WHERE user_uid=:u AND target_type != '' AND deleted_at IS NULL",
u=user_uid,
)
)[0]["n"]
pagination = build_pagination(page, total, per_page)
offset = (pagination["page"] - 1) * pagination["per_page"]
rows = db.query(
"SELECT * FROM attachments "
"WHERE user_uid=:u AND target_type != '' AND deleted_at IS NULL "
"ORDER BY created_at DESC LIMIT :limit OFFSET :offset",
u=user_uid,
limit=pagination["per_page"],
offset=offset,
)
items = []
for row in rows:
item = _row_to_attachment(row)
item["target_url"] = resolve_object_url(item["target_type"], item["target_uid"])
items.append(item)
return items, pagination
def get_deleted_media(page: int = 1, per_page: int = 24) -> tuple:
if "attachments" not in db.tables:
return [], build_pagination(page, 0, per_page)
from devplacepy.attachments import _row_to_attachment
total = list(
db.query("SELECT COUNT(*) AS n FROM attachments WHERE deleted_at IS NOT NULL")
)[0]["n"]
pagination = build_pagination(page, total, per_page)
offset = (pagination["page"] - 1) * pagination["per_page"]
rows = db.query(
"SELECT * FROM attachments WHERE deleted_at IS NOT NULL "
"ORDER BY deleted_at DESC LIMIT :limit OFFSET :offset",
limit=pagination["per_page"],
offset=offset,
)
usernames = {u["uid"]: u["username"] for u in get_table("users").find()}
items = []
for row in rows:
item = _row_to_attachment(row)
item["target_url"] = resolve_object_url(item["target_type"], item["target_uid"])
item["deleted_at"] = row.get("deleted_at", "")
item["uploader"] = usernames.get(row.get("user_uid"), "unknown")
items.append(item)
return items, pagination
def get_follow_counts(user_uid: str) -> dict:
if "follows" not in db.tables:
return {"followers": 0, "following": 0}
follows = get_table("follows")
return {
"followers": follows.count(following_uid=user_uid, deleted_at=None),
"following": follows.count(follower_uid=user_uid, deleted_at=None),
}
def get_follow_list(
user_uid: str, mode: str, page: int = 1, per_page: int = 25
) -> tuple:
if "follows" not in db.tables:
return [], build_pagination(page, 0, per_page)
follows = get_table("follows")
key = "following_uid" if mode == "followers" else "follower_uid"
other = "follower_uid" if mode == "followers" else "following_uid"
total = follows.count(deleted_at=None, **{key: user_uid})
pagination = build_pagination(page, total, per_page)
offset = (pagination["page"] - 1) * pagination["per_page"]
rows = list(
follows.find(
order_by=["-created_at"],
_limit=pagination["per_page"],
_offset=offset,
deleted_at=None,
**{key: user_uid},
)
)
users_map = get_users_by_uids([row[other] for row in rows])
people = []
for row in rows:
person = users_map.get(row[other])
if person:
people.append(
{
"uid": person["uid"],
"username": person["username"],
"bio": (person.get("bio") or "")[:140],
"followed_at": row.get("created_at"),
}
)
return people, pagination
def get_following_among(follower_uid: str, target_uids: list) -> set:
if not follower_uid or not target_uids or "follows" not in db.tables:
return set()
placeholders, params = _in_clause(target_uids)
params["f"] = follower_uid
rows = db.query(
f"SELECT following_uid FROM follows WHERE follower_uid = :f AND following_uid IN ({placeholders}) AND deleted_at IS NULL",
**params,
)
return {row["following_uid"] for row in rows}
def get_daily_topic():
if "news" in db.tables:
article = db["news"].find_one(
status="published", deleted_at=None, order_by=["-synced_at"]
)
if article:
desc = (article.get("description") or "")[:200] or (
article.get("content") or ""
)[:200]
return {
"title": article.get("title", ""),
"summary": desc,
"slug": article.get("slug", ""),
"url": article.get("url", ""),
"image_url": article.get("image_url", ""),
}
return {
"title": "Welcome to DevPlace",
"summary": "Stay tuned for the latest dev news.",
}
def get_featured_news(limit=5):
if "news" not in db.tables:
return []
from devplacepy.utils import time_ago
rows = list(
db["news"].find(
show_on_landing=1, deleted_at=None, order_by=["-synced_at"], _limit=limit
)
)
articles = []
for article in rows:
summary = (article.get("description") or "")[:120] or (
article.get("content") or ""
)[:120]
articles.append(
{
"title": article.get("title", ""),
"summary": summary,
"slug": article.get("slug", ""),
"url": article.get("url", ""),
"source_name": article.get("source_name", ""),
"time_ago": time_ago(article["synced_at"])
if article.get("synced_at")
else "",
}
)
return articles
def _ds_now() -> str:
return datetime.now(timezone.utc).isoformat()
def create_deepsearch_session(
uid: str,
owner_kind: str,
owner_id: str,
query: str,
depth: int,
max_pages: int,
collection: str,
) -> None:
get_table("deepsearch_sessions").insert(
{
"uid": uid,
"owner_kind": owner_kind,
"owner_id": owner_id,
"query": query,
"status": "pending",
"depth": depth,
"max_pages": max_pages,
"score": 0,
"confidence": 0.0,
"source_diversity": 0.0,
"page_count": 0,
"chunk_count": 0,
"collection": collection,
"summary": "",
"created_at": _ds_now(),
"completed_at": "",
"deleted_at": None,
"deleted_by": None,
}
)
def update_deepsearch_session(uid: str, fields: dict) -> None:
if "deepsearch_sessions" not in db.tables:
return
payload = dict(fields)
payload["uid"] = uid
get_table("deepsearch_sessions").update(payload, ["uid"])
def get_deepsearch_session(uid: str) -> dict | None:
if "deepsearch_sessions" not in db.tables:
return None
return get_table("deepsearch_sessions").find_one(uid=uid, deleted_at=None)
def add_deepsearch_message(
uid: str, session_uid: str, role: str, content: str, citations: str = ""
) -> None:
get_table("deepsearch_messages").insert(
{
"uid": uid,
"session_uid": session_uid,
"role": role,
"content": content,
"citations": citations,
"created_at": _ds_now(),
"deleted_at": None,
"deleted_by": None,
}
)
def get_deepsearch_messages(session_uid: str, limit: int = 50) -> list[dict]:
if "deepsearch_messages" not in db.tables:
return []
return list(
get_table("deepsearch_messages").find(
session_uid=session_uid,
deleted_at=None,
order_by=["created_at"],
_limit=limit,
)
)
def get_cached_deepsearch_url(url_hash: str) -> dict | None:
if "deepsearch_url_cache" not in db.tables:
return None
return get_table("deepsearch_url_cache").find_one(url_hash=url_hash)
def upsert_deepsearch_url_cache(
url_hash: str,
url: str,
title: str,
content_hash: str,
status: int,
byte_size: int,
) -> None:
table = get_table("deepsearch_url_cache")
existing = table.find_one(url_hash=url_hash)
row = {
"url_hash": url_hash,
"url": url,
"title": title,
"content_hash": content_hash,
"status": status,
"byte_size": byte_size,
"fetched_at": _ds_now(),
}
if existing:
row["uid"] = existing["uid"]
table.update(row, ["uid"])
else:
from devplacepy.utils import generate_uid
row["uid"] = generate_uid()
table.insert(row)
Reference in New Issue View Git Blame Copy Permalink