Database Module

The database module provides an interface for storing and retrieving conference papers using SQLAlchemy, supporting both SQLite and PostgreSQL backends.

Overview

The DatabaseManager class handles all database operations including:

• Creating and managing the database schema

• Adding papers and paper metadata

• Searching and filtering papers

• Storing embeddings metadata, clustering caches, validation data, and evaluation results

Class Reference

This module provides functionality to load JSON data into a SQL database. Supports both SQLite and PostgreSQL backends via SQLAlchemy.

exception abstracts_explorer.database.DatabaseError

Bases: Exception

Exception raised for database operations.

exception abstracts_explorer.database.EmbeddingModelConflictError(local_model, remote_model)

Bases: DatabaseError

Raised when the embedding model in imported data differs from the local database.

Variables:

• local_model (str) – Embedding model currently in the local database.

• remote_model (str) – Embedding model in the data being imported.

__init__(local_model, remote_model)

abstracts_explorer.database.normalize_model_name(name)

Normalize an embedding model name for comparison.

Strips a leading alias- prefix (case-insensitive) so that, e.g., alias-qwen3-embeddings-8b is considered identical to qwen3-embeddings-8b.

Parameters:

name (str) – Embedding model name.

Returns:

Normalized model name.

Return type:

str

class abstracts_explorer.database.DatabaseManager

Bases: object

Manager for SQL database operations using SQLAlchemy.

Supports SQLite and PostgreSQL backends through SQLAlchemy connection URLs. Database configuration is read from the config file (PAPER_DB variable).

Variables:

• database_url (str) – SQLAlchemy database URL from configuration.

• engine (Engine or None) – SQLAlchemy engine instance.

• SessionLocal (sessionmaker or None) – SQLAlchemy session factory.

• _session (Session or None) – Internal active database session.

• session (Session) – Lazy session property. The first access connects to the database and creates any missing tables.

Examples

>>> # Database configuration comes from config file
>>> db = DatabaseManager()
>>> db.get_paper_count()  # connects and creates tables automatically
0
>>> db.close()

__init__()

Initialize the DatabaseManager.

Reads database configuration from the config file.

property session: Session

Return the active database session, connecting lazily if needed.

The first access calls connect() and create_tables() so that the database schema is available automatically.

Returns:

Active SQLAlchemy session.

Return type:

Session

Raises:

DatabaseError – If connecting to the database or creating the schema fails.

connect()

Connect to the database.

Does nothing if already connected. Creates the database file if it does not exist (SQLite only).

Raises:

DatabaseError – If connection fails.

Return type:

None

close()

Close the database connection.

Does nothing if not connected.

Return type:

None

__enter__()

Context manager entry.

__exit__(exc_type, exc_val, exc_tb)

Context manager exit.

static compute_uid(title, original_id, conference, year)

Compute a deterministic paper UID from its identifying fields.

The UID is a 16-character hex string derived from a SHA-256 hash of the concatenated title, original_id, conference and year.

Parameters:

• title (str) – Paper title.

• original_id (Union[str, int, None]) – Original paper ID from the source (e.g., OpenReview ID).

• conference (str) – Conference name (e.g., “NeurIPS”).

• year (int) – Conference year.

Returns:

16-character hex UID.

Return type:

str

Examples

>>> DatabaseManager.compute_uid("My Paper", "abc123", "NeurIPS", 2025)
'a1b2c3d4e5f67890'

create_tables()

Create database tables for papers and embeddings metadata.

Creates the following tables: - papers: Main table for paper information with lightweight ML4PS schema - embeddings_metadata: Metadata about embeddings (model used, creation date) - clustering_cache: Cache for clustering results

This method is idempotent - it can be called multiple times without error. Tables are only created if they don’t already exist.

Raises:

DatabaseError – If table creation fails.

Return type:

None

add_paper(paper)

Add a single paper to the database.

Parameters:

paper (LightweightPaper) – Validated paper object to insert.

Returns:

The UID of the inserted paper, or None if paper was skipped (duplicate).

Return type:

Optional[str]

Raises:

DatabaseError – If insertion fails.

Examples

>>> from abstracts_explorer.plugin import LightweightPaper
>>> db = DatabaseManager()
>>> with db:
...     db.create_tables()
...     paper = LightweightPaper(
...         title="Test Paper",
...         authors=["John Doe"],
...         abstract="Test abstract",
...         session="Session 1",
...         poster_position="P1",
...         year=2025,
...         conference="NeurIPS"
...     )
...     paper_uid = db.add_paper(paper)
>>> print(f"Inserted paper with UID: {paper_uid}")

add_papers(papers)

Add multiple papers to the database in a batch.

Parameters:

papers (List[LightweightPaper]) – List of validated paper objects to insert.

Returns:

Number of papers successfully inserted (excludes duplicates).

Return type:

int

Raises:

DatabaseError – If batch insertion fails.

Examples

>>> from abstracts_explorer.plugin import LightweightPaper
>>> db = DatabaseManager()
>>> with db:
...     db.create_tables()
...     papers = [
...         LightweightPaper(
...             title="Paper 1",
...             authors=["Author 1"],
...             abstract="Abstract 1",
...             session="Session 1",
...             poster_position="P1",
...             year=2025,
...             conference="NeurIPS"
...         ),
...         LightweightPaper(
...             title="Paper 2",
...             authors=["Author 2"],
...             abstract="Abstract 2",
...             session="Session 2",
...             poster_position="P2",
...             year=2025,
...             conference="NeurIPS"
...         )
...     ]
...     count = db.add_papers(papers)
>>> print(f"Inserted {count} papers")

donate_validation_data(paper_priorities)

Store donated paper rating data for validation purposes.

This method accepts anonymized paper ratings from users and stores them in the validation_data table for improving the service.

Parameters:

paper_priorities (Dict[str, Dict[str, Any]]) – Dictionary mapping paper UIDs to priority data. Each priority data dict must contain: - priority (int): Rating value - searchTerm (str, optional): Search term associated with the rating

Returns:

Number of papers successfully donated

Return type:

int

Raises:

• ValueError – If paper_priorities is empty or contains invalid data format

• DatabaseError – If database operation fails

Examples

>>> db = DatabaseManager()
>>> with db:
...     priorities = {
...         "abc123": {"priority": 5, "searchTerm": "machine learning"},
...         "def456": {"priority": 4, "searchTerm": "deep learning"}
...     }
...     count = db.donate_validation_data(priorities)
...     print(f"Donated {count} papers")

donate_chat_transcript(rating, transcript)

Store a donated chat transcript with thumbs up/down feedback.

This method accepts an anonymized chat transcript and a rating from the user and stores it for improving the chat system.

Parameters:

• rating (str) – User feedback rating, must be ‘up’ or ‘down’.

• transcript (List[Dict[str, str]]) – List of message dicts, each with ‘role’ and ‘text’ keys.

Returns:

ID of the stored donation entry.

Return type:

int

Raises:

• ValueError – If rating is invalid or transcript is empty/malformed.

• DatabaseError – If database operation fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     transcript = [
...         {"role": "user", "text": "What papers discuss transformers?"},
...         {"role": "assistant", "text": "Here are some relevant papers..."}
...     ]
...     donation_id = db.donate_chat_transcript("up", transcript)

get_chat_donations(limit=None, rating=None, offset=0)

Retrieve donated chat transcripts from the database.

Parameters:

• limit (Optional[int]) – Maximum number of entries to return. Returns all if None.

• rating (Optional[str]) – Filter by rating (‘up’ or ‘down’). Returns all ratings if None.

• offset (int) – Number of entries to skip for pagination (default: 0).

Returns:

List of donation dicts, each containing: - id (int): Entry ID. - rating (str): ‘up’ or ‘down’. - transcript (list): Parsed list of message dicts. - donated_at (datetime): Donation timestamp.

Return type:

List[Dict[str, Any]]

Raises:

DatabaseError – If the database operation fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     donations = db.get_chat_donations(limit=10, rating='up')

get_chat_donation_stats()

Get summary statistics for donated chat transcripts.

Returns:

Statistics dict containing: - total (int): Total number of donations. - up (int): Number of thumbs-up donations. - down (int): Number of thumbs-down donations. - avg_turns (float): Average number of turns per transcript.

Return type:

Dict[str, Any]

Raises:

DatabaseError – If the database operation fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     stats = db.get_chat_donation_stats()
...     print(f"Total donations: {stats['total']}")

get_validation_data(limit=None, offset=0)

Retrieve donated validation (interesting-paper) data from the database.

Parameters:

• limit (Optional[int]) – Maximum number of entries to return. Returns all if None.

• offset (int) – Number of entries to skip for pagination (default: 0).

Returns:

List of validation data dicts, each containing: - id (int): Entry ID. - paper_uid (str): Paper UID. - priority (int): Priority rating. - search_term (str or None): Associated search term. - donated_at (datetime): Donation timestamp.

Return type:

List[Dict[str, Any]]

Raises:

DatabaseError – If the database operation fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     data = db.get_validation_data(limit=20)

get_validation_data_stats()

Get summary statistics for donated validation (interesting-paper) data.

Returns:

Statistics dict containing: - total (int): Total number of donated paper ratings. - unique_papers (int): Number of distinct paper UIDs. - avg_priority (float): Average priority rating. - priority_distribution (dict): Count per priority value.

Return type:

Dict[str, Any]

Raises:

DatabaseError – If the database operation fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     stats = db.get_validation_data_stats()
...     print(f"Total data donations: {stats['total']}")

delete_chat_donations(ids=None)

Delete donated chat transcripts.

Parameters:

ids (Optional[List[int]]) – List of donation IDs to delete. If None, deletes all donations.

Returns:

Number of donations deleted.

Return type:

int

Raises:

DatabaseError – If the database operation fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     deleted = db.delete_chat_donations()
...     print(f"Deleted {deleted} donations")

delete_validation_data(ids=None)

Delete donated validation (interesting-paper) data.

Parameters:

ids (Optional[List[int]]) – List of entry IDs to delete. If None, deletes all validation data.

Returns:

Number of entries deleted.

Return type:

int

Raises:

DatabaseError – If the database operation fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     deleted = db.delete_validation_data()
...     print(f"Deleted {deleted} entries")

get_paper(where_filters=None, limit=100)

Retrieve papers matching where_filters.

This is the single authoritative paper-search method. All other paper retrieval methods delegate to it.

Parameters:

• where_filters (Optional[Dict[str, Any]]) – Mapping of Paper column names to filter values. See _build_where_conditions() for the supported value forms.

• limit (int) – Maximum number of results. 0 or None means no limit.

Returns:

Matching papers as dictionaries.

Return type:

List[Dict[str, Any]]

Raises:

DatabaseError – If the query fails.

get_paper_count(where_filters=None)

Count papers in the database, optionally filtered.

Parameters:

where_filters (Optional[Dict[str, Any]]) – Mapping of Paper column names to filter values.

Returns:

Number of matching papers.

Return type:

int

Raises:

DatabaseError – If query fails.

get_paper_by_uid(uid)

Retrieve a paper by its UID.

Parameters:

uid (str) – UID of the paper to retrieve.

Returns:

Paper data as a dictionary, or None if not found.

Return type:

Optional[Dict[str, Any]]

Raises:

DatabaseError – If query fails.

get_papers_by_uids(uids)

Retrieve multiple papers by their UIDs in a single query.

Parameters:

uids (List[str]) – UIDs of the papers to retrieve.

Returns:

Paper data dictionaries for all matching UIDs. The order follows the input uids list, and unknown UIDs are omitted.

Return type:

List[Dict[str, Any]]

Raises:

DatabaseError – If the query fails.

get_paper_by_original_id_or_uid(paper_id)

Retrieve a paper by its UID or original_id (whichever matches first).

Parameters:

paper_id (str) – Value to match against the uid or original_id column.

Returns:

Fully formatted paper data dictionary, or None if not found.

Return type:

Optional[Dict[str, Any]]

Raises:

DatabaseError – If the database query fails.

SEARCHABLE_FIELDS: set = {'abstract', 'authors', 'award', 'conference', 'endtime', 'keywords', 'original_id', 'paper_pdf_url', 'poster_image_url', 'poster_position', 'room_name', 'session', 'starttime', 'title', 'url', 'year'}

Paper model column names that can be used as field:"value" filters in search queries. Internal columns (uid, created_at) are excluded because they are not meaningful search targets for users.

FIELD_ALIASES: Dict[str, str] = {'author': 'authors'}

Aliases for field names in search queries. An alias is transparently resolved to the canonical column name before applying the filter.

search_papers(keyword=None, field_filters=None, sessions=None, years=None, conferences=None, limit=100)

Search papers in the SQL database.

This method handles database search requests with keyword, field-specific, session, year and conference filters. It delegates the actual query to get_paper().

Parameters:

• keyword (Optional[str]) – Keyword to search in title, abstract, or keywords fields.

• field_filters (Optional[Dict[str, str]]) – Mapping of Paper column names to search values. Each entry adds a case-insensitive substring condition on the corresponding column (e.g. {"authors": "Smith", "award": "Best Paper"}).

• sessions (Optional[List[str]]) – List of sessions to filter by (matches ANY).

• years (Optional[List[int]]) – List of years to filter by (matches ANY).

• conferences (Optional[List[str]]) – List of conferences to filter by (matches ANY).

• limit (int) – Maximum number of results to return.

Returns:

Matching papers as dictionaries.

Return type:

List[Dict[str, Any]]

Raises:

DatabaseError – If search fails.

static parse_field_filters(query)

Parse field-specific filters from a search query string.

Extracts all field:"value" patterns from the query where field is a column name of the Paper model (or a recognised alias; see FIELD_ALIASES). Unrecognised field names are left in the query text as-is.

Parameters:

query (str) – The raw search query, e.g. 'authors:"John Smith" award:"Best Paper" transformers'.

Returns:

A tuple (field_filters, remaining_query) where field_filters maps canonical column names to their search values and remaining_query is the query with recognised filters removed.

Return type:

tuple

Examples

>>> DatabaseManager.parse_field_filters('authors:"John Smith" transformers')
({'authors': 'John Smith'}, 'transformers')
>>> DatabaseManager.parse_field_filters('author:"John Smith" transformers')
({'authors': 'John Smith'}, 'transformers')
>>> DatabaseManager.parse_field_filters('Author:"John Smith" transformers')
({'authors': 'John Smith'}, 'transformers')
>>> DatabaseManager.parse_field_filters('transformers')
({}, 'transformers')
>>> DatabaseManager.parse_field_filters('award:"Best Paper" authors:"Doe"')
({'award': 'Best Paper', 'authors': 'Doe'}, '')

search_papers_keyword(query, limit=10, sessions=None, years=None, conferences=None, *, use_embeddings=False, distance_threshold=1.1, embeddings_manager=None)

Perform a user-query search, optionally using semantic embeddings.

This is the main search entry point for the web UI and CLI. It parses field:"value" filters from query, converts the remaining filter arguments into structured where filters, and then delegates either to search_papers() or search_papers_semantic().

Parameters:

• query (str) – Search query text. May include field:"value" filters.

• limit (int) – Maximum number of results, by default 10.

• sessions (Optional[List[str]]) – Filter by paper sessions.

• years (Optional[List[int]]) – Filter by publication years.

• conferences (Optional[List[str]]) – Filter by conference names.

• use_embeddings (bool) – Use semantic search instead of SQL keyword search.

• distance_threshold (float) – Maximum embedding distance for semantic results.

Returns:

List of paper dictionaries.

Return type:

List[Dict[str, Any]]

Raises:

DatabaseError – If the search fails.

search_papers_semantic(query, limit=10, sessions=None, years=None, conferences=None, distance_threshold=1.1, field_filters=None, where=None, embeddings_manager=None)

Perform semantic paper search using embeddings.

Field filters parsed from the query are applied through the SQL database. The remaining query text is used for the embedding similarity search. Author-name matches are returned first when the query text matches an author.

Parameters:

• query (str) – Semantic query text (field filters already removed when field_filters is supplied).

• limit (int) – Maximum number of results, by default 10.

• sessions (Optional[List[str]]) – Filter by paper sessions.

• years (Optional[List[int]]) – Filter by publication years.

• conferences (Optional[List[str]]) – Filter by conference names.

• distance_threshold (float) – Maximum embedding distance for results, by default 1.1.

• field_filters (Optional[Dict[str, str]]) – Field filters to apply in SQL. When omitted, filters are parsed from query.

• where (Optional[Dict[str, Any]]) – Additional ChromaDB metadata filter. Merged with the session/year/conference filters.

Returns:

Paper dictionaries ordered by relevance.

Return type:

List[Dict[str, Any]]

Raises:

DatabaseError – If the search fails.

get_stats(year=None, conference=None)

Get database statistics, optionally filtered by year and conference.

Parameters:

• year (Optional[int]) – Filter by specific year

• conference (Optional[str]) – Filter by specific conference

Returns:

Statistics dictionary with: - total_papers: int - Number of papers matching filters - year: int or None - Filter year if provided - conference: str or None - Filter conference if provided

Return type:

Dict[str, Any]

Examples

>>> stats = db.get_stats()
>>> print(f"Total papers: {stats['total_papers']}")

>>> stats_2024 = db.get_stats(year=2024)
>>> print(f"Papers in 2024: {stats_2024['total_papers']}")

get_conference_years_from_db()

Return a mapping of each conference to the years that have papers in the database.

Years are sorted in descending order (most recent first).

Returns:

Mapping of conference name to list of years (descending) that have at least one paper in the database. Returns an empty dict when the database is empty.

Return type:

Dict[str, List[int]]

Raises:

DatabaseError – If the query fails.

Examples

>>> db = DatabaseManager()
>>> mapping = db.get_conference_years_from_db()
>>> print(mapping)
{'NeurIPS': [2025, 2024], 'ICLR': [2024]}

confirm_default_conference_year_exists(configured_conference, configured_year)

Confirm that the configured default conference/year exists in the database.

Unlike the previous fallback logic, this method never guesses an alternative. If the configured default is missing, it returns a structured error describing available data and how to fix the problem.

Parameters:

• configured_conference (str) – Conference name from the application configuration.

• configured_year (Optional[int]) – Year from the application configuration. 0 or None means that no year is configured; only the conference is validated.

Returns:

Dictionary with keys:

• conference : str or None – canonical configured conference, or None when it is unavailable.

• year : int or None – confirmed configured year, or None.

• error : dict or None – error details when the default is unavailable. Contains message, available_conferences, available_years and guidance.

Return type:

Dict[str, Any]

resolve_conference_name(conference)

Resolve a conference name to the canonical form stored in the database.

Performs a case-insensitive match against conference names already present in the database. If no database match is found, falls back to a case-insensitive match against the conference_name attribute of every registered downloader plugin. If neither lookup succeeds the original conference string is returned unchanged.

This is the single authoritative place where conference-name normalization must happen. CLI commands should call this method once at the entry point of each command and then work with the returned canonical name for all subsequent operations.

Parameters:

conference (str) – Conference name as supplied by the caller. May differ in case or spelling from the form stored in the database (e.g. ml4ps vs. ML4PS).

Returns:

The conference name exactly as it appears in the database (first match), or exactly as defined by the first matching plugin, or the input string if no match is found.

Return type:

str

Examples

>>> with DatabaseManager() as db:
...     db.create_tables()
...     canonical = db.resolve_conference_name("ml4ps")
...     # Returns "ML4PS" if that form is stored in the DB

resolve_conference_for_url(url_path)

Resolve a URL path segment to a conference, checking data availability.

Combines plugin-based name resolution with a database data check. Returns a result dict describing the outcome:

• found with data: {"conference": "<name>", "error": None}

• found without data: {"conference": None, "error": {"message": "...", "available_conferences": [...]}}

• not found: {"conference": None, "error": {"message": "...", "available_conferences": [...]}}

Parameters:

url_path (str) – URL path segment (e.g. "neurips", "ICLR").

Returns:

Dictionary with keys "conference" (str or None) and "error" (dict or None).

Return type:

dict

Examples

>>> with DatabaseManager() as db:
...     result = db.resolve_conference_for_url("neurips")
...     if result["conference"]:
...         print(f"Found: {result['conference']}")

get_sessions(conference=None, year=None)

Get distinct session names from the database.

Parameters:

• conference (Optional[str]) – Filter sessions to only those belonging to this conference.

• year (Optional[int]) – Filter sessions to only those belonging to this year.

Returns:

Sorted list of distinct non-empty session names.

Return type:

List[str]

Raises:

DatabaseError – If query fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     sessions = db.get_sessions()
>>> print(sessions)
['Session 1', 'Session 2', ...]
>>> sessions = db.get_sessions(conference="NeurIPS", year=2025)

get_conferences(year=None)

Get distinct conference names from the database.

Parameters:

year (Optional[int]) – Filter conferences to only those that have papers for this year.

Returns:

Sorted list of distinct non-empty conference names.

Return type:

List[str]

Raises:

DatabaseError – If query fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     conferences = db.get_conferences()
>>> print(conferences)
['ICLR', 'NeurIPS']
>>> conferences = db.get_conferences(year=2025)

get_years(conference=None)

Get distinct years from the database, sorted descending (most recent first).

Parameters:

conference (Optional[str]) – Filter years to only those that have papers for this conference.

Returns:

Sorted list (descending) of distinct years.

Return type:

List[int]

Raises:

DatabaseError – If query fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     years = db.get_years()
>>> print(years)
[2025, 2024, 2023]
>>> years = db.get_years(conference="NeurIPS")

get_embedding_model()

Get the embedding model used for the current embeddings.

Returns:

Name of the embedding model, or None if not set.

Return type:

Optional[str]

Raises:

DatabaseError – If query fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     model = db.get_embedding_model()
>>> print(model)
'text-embedding-qwen3-embedding-4b'

set_embedding_model(model_name)

Set the embedding model used for embeddings.

This stores or updates the embedding model metadata. If a record exists, it updates the model and timestamp. Otherwise, it creates a new record.

Parameters:

model_name (str) – Name of the embedding model.

Raises:

DatabaseError – If update fails.

Return type:

None

Examples

>>> db = DatabaseManager()
>>> with db:
...     db.set_embedding_model("text-embedding-qwen3-embedding-4b")

get_clustering_cache(embedding_model, clustering_method, n_clusters=None, clustering_params=None, reduction_method=None, n_components=None, conference=None, year=None)

Get cached clustering results matching the parameters.

When reduction_method and n_components are provided, only entries that match exactly (including the reduction method) are returned. When they are omitted (None), the reduction method is ignored and the most recent entry matching the clustering parameters is returned.

Parameters:

• embedding_model (str) – Name of the embedding model.

• clustering_method (str) – Clustering algorithm used.

• n_clusters (Optional[int]) – Number of clusters to match. When None the query does not filter by this column.

• clustering_params (Optional[Dict[str, Any]]) – Additional clustering parameters (e.g., distance_threshold, eps).

• reduction_method (Optional[str]) – Dimensionality reduction method. When provided, the query requires an exact match on this column.

• n_components (Optional[int]) – Number of components after reduction. When provided, the query requires an exact match on this column.

• conference (Optional[str]) – Conference name to filter by. None matches entries that have no conference set.

• year (Optional[int]) – Conference year to filter by. None matches entries that have no year set.

Returns:

Cached clustering results as dictionary, or None if not found.

Return type:

Optional[Dict[str, Any]]

Raises:

DatabaseError – If query fails.

save_clustering_cache(embedding_model, reduction_method, n_components, clustering_method, results, n_clusters=None, clustering_params=None, conference=None, year=None)

Save clustering results to cache.

The full results including visualization coordinates are stored. The reduction_method and n_components are stored so that an exact-match lookup can return cached points directly. When only the reduction method changes, the clustering results are reused and only the reduction is re-applied.

Parameters:

• embedding_model (str) – Name of the embedding model.

• reduction_method (str) – Dimensionality reduction method.

• n_components (int) – Number of components after reduction.

• clustering_method (str) – Clustering algorithm used.

• results (Dict[str, Any]) – Clustering results to cache (full results including points).

• n_clusters (Optional[int]) – Number of clusters. When None, the actual count is extracted from results["statistics"]["n_clusters"].

• clustering_params (Optional[Dict[str, Any]]) – Additional clustering parameters.

• conference (Optional[str]) – Conference name this entry is scoped to.

• year (Optional[int]) – Conference year this entry is scoped to.

Raises:

DatabaseError – If save fails.

Return type:

None

delete_papers_by_conference_year(conference, year)

Delete all papers for a specific conference and year combination.

Parameters:

• conference (str) – Conference name (exact, as stored in the database).

• year (int) – Conference year.

Returns:

Number of papers deleted.

Return type:

int

Raises:

DatabaseError – If deletion fails.

delete_clustering_cache_by_conference_year(conference, year)

Delete all clustering cache entries for a specific conference and year.

Parameters:

• conference (str) – Conference name (exact, as stored in the database).

• year (int) – Conference year.

Returns:

Number of cache entries deleted.

Return type:

int

Raises:

DatabaseError – If deletion fails.

count_clustering_cache_by_conference_year(conference, year)

Count clustering cache entries for a specific conference and year.

Parameters:

• conference (str) – Conference name (exact, as stored in the database).

• year (int) – Conference year.

Returns:

Number of cache entries found.

Return type:

int

Raises:

DatabaseError – If the query fails.

clear_clustering_cache(embedding_model=None)

Clear clustering cache, optionally filtered by embedding model.

This is useful when embeddings change or cache becomes stale.

Parameters:

embedding_model (Optional[str]) – If provided, only clear cache for this embedding model. If None, clear all cache entries.

Returns:

Number of cache entries deleted.

Return type:

int

Raises:

DatabaseError – If deletion fails.

update_clustering_cache_embedding_model(old_model, new_model)

Update the embedding model name in all clustering cache entries.

Renames every ClusteringCache row whose embedding_model matches old_model so that it refers to new_model instead. This keeps cached clustering results usable after the embedding model metadata is renamed (e.g. via update_embedding_model_metadata.py).

Parameters:

• old_model (str) – Current embedding model name stored in the cache.

• new_model (str) – New embedding model name to write.

Returns:

Number of cache entries updated.

Return type:

int

Raises:

DatabaseError – If the update fails.

Examples

>>> db = DatabaseManager()
>>> with db:
...     count = db.update_clustering_cache_embedding_model(
...         "old-model", "new-model"
...     )

get_hierarchical_label_cache(embedding_model, linkage='ward')

Get cached hierarchical labels for agglomerative clustering.

Hierarchical labels are independent of the number of clusters and the distance threshold, so they are reused for all agglomerative clustering settings that share the same embedding model and linkage.

Parameters:

• embedding_model (str) – Name of the embedding model.

• linkage (str) – Agglomerative linkage method (default: "ward").

Returns:

Mapping of {node_id: label} (integer keys), or None if no entry is found.

Return type:

Optional[Dict[int, str]]

Raises:

DatabaseError – If query fails.

save_hierarchical_label_cache(embedding_model, labels, linkage='ward')

Save hierarchical cluster labels to cache.

Parameters:

• embedding_model (str) – Name of the embedding model.

• labels (Dict[int, str]) – Mapping of {node_id: label} to store.

• linkage (str) – Agglomerative linkage method (default: "ward").

Raises:

DatabaseError – If save fails.

Return type:

None

add_eval_qa_pair(conversation_id, turn_number, query, expected_answer, tool_name=None, source_info=None)

Insert a single evaluation Q/A pair.

Parameters:

• conversation_id (str) – Identifier grouping turns in a conversation.

• turn_number (int) – Position within the conversation (0 = first).

• query (str) – The user query text.

• expected_answer (str) – The expected/reference answer.

• tool_name (Optional[str]) – MCP tool expected to be invoked.

• source_info (Optional[str]) – JSON metadata about how the pair was generated.

Returns:

Primary key of the inserted row.

Return type:

int

Raises:

DatabaseError – If insertion fails.

get_eval_qa_pairs(verified_only=False, tool_name=None, conversation_id=None, limit=None, offset=0)

Retrieve evaluation Q/A pairs with optional filters.

Parameters:

• verified_only (bool) – If True, return only pairs with verified == 1.

• tool_name (Optional[str]) – Filter by expected MCP tool name.

• conversation_id (Optional[str]) – Filter by conversation.

• limit (Optional[int]) – Maximum number of pairs to return.

• offset (int) – Number of rows to skip (for pagination).

Returns:

Matching Q/A pairs as dictionaries.

Return type:

List[Dict[str, Any]]

Raises:

DatabaseError – If query fails.

get_eval_qa_pair_count(verified_only=False)

Count evaluation Q/A pairs.

Parameters:

verified_only (bool) – If True, count only verified pairs.

Returns:

Number of matching pairs.

Return type:

int

Raises:

DatabaseError – If query fails.

update_eval_qa_pair(pair_id, **fields)

Update fields on an existing Q/A pair.

Parameters:

• pair_id (int) – Primary key of the pair to update.

• **fields – Keyword arguments mapping column names to new values. Supported keys: query, expected_answer, tool_name, verified, source_info.

Returns:

True if a row was updated, False if the pair was not found.

Return type:

bool

Raises:

DatabaseError – If update fails.

delete_eval_qa_pair(pair_id)

Delete an evaluation Q/A pair by ID.

Parameters:

pair_id (int) – Primary key of the pair to delete.

Returns:

True if a row was deleted, False if pair was not found.

Return type:

bool

Raises:

DatabaseError – If deletion fails.

delete_verified_eval_qa_pairs()

Delete all verified (accepted) evaluation Q/A pairs.

Returns:

Number of pairs deleted.

Return type:

int

Raises:

DatabaseError – If deletion fails.

delete_eval_results(run_id=None)

Delete stored evaluation results, optionally filtered to a single run.

Parameters:

run_id (Optional[str]) – If supplied, only results for this run are deleted. If None, all stored results are deleted.

Returns:

Number of rows deleted.

Return type:

int

Raises:

DatabaseError – If deletion fails.

add_eval_result(run_id, qa_pair_id, actual_answer=None, actual_tool_name=None, answer_score=None, tool_correct=None, latency_ms=None, error=None, judge_reasoning=None)

Insert a single evaluation result.

Parameters:

• run_id (str) – Identifier for the evaluation run.

• qa_pair_id (int) – ID of the evaluated Q/A pair.

• actual_answer (Optional[str]) – Answer produced by the RAG system.

• actual_tool_name (Optional[str]) – MCP tool actually invoked.

• answer_score (Optional[float]) – LLM-judged quality score (1–5).

• tool_correct (Optional[int]) – 1 if the correct tool was used, 0 otherwise.

• latency_ms (Optional[int]) – Query latency in milliseconds.

• error (Optional[str]) – Error message if the query failed.

• judge_reasoning (Optional[str]) – LLM judge’s reasoning for the score.

Returns:

Primary key of the inserted row.

Return type:

int

Raises:

DatabaseError – If insertion fails.

get_eval_results(run_id=None, limit=None, offset=0)

Retrieve evaluation results with optional run filter.

Parameters:

• run_id (Optional[str]) – Filter by evaluation run. If None, return results from all runs.

• limit (Optional[int]) – Maximum number of results to return.

• offset (int) – Number of rows to skip.

Returns:

Evaluation results as dictionaries.

Return type:

List[Dict[str, Any]]

Raises:

DatabaseError – If query fails.

get_eval_run_ids()

Return distinct evaluation run IDs ordered by run time, oldest first.

The ordering is determined by the minimum created_at timestamp of all results in each run, so the most recent run appears last.

Returns:

Distinct run IDs ordered chronologically (oldest to newest).

Return type:

List[str]

Raises:

DatabaseError – If query fails.

get_eval_run_summary(run_id)

Compute summary statistics for an evaluation run.

Parameters:

run_id (str) – The evaluation run identifier.

Returns:

Dictionary with keys:

• total : int – number of evaluated pairs

• avg_score : float or None – mean answer quality score

• tool_accuracy : float or None – fraction of correct tool selections

• avg_latency_ms : float or None – mean latency

• error_count : int – number of queries that produced errors

• run_date : datetime or None – timestamp of the first result in the run

Return type:

Dict[str, Any]

Raises:

DatabaseError – If query fails.

export_papers_to_sqlite(output_path, conference, year)

Export papers for a given conference and year to a standalone SQLite file.

The exported file includes hierarchical label cache and embeddings metadata rows. Clustering cache is not included — it is exported separately via export_clustering_cache_to_json().

Parameters:

• output_path (Path) – Destination path for the SQLite file.

• conference (str) – Conference name to export.

• year (int) – Year to export.

Returns:

Number of papers exported.

Return type:

int

Raises:

DatabaseError – If the export fails or no papers are found.

import_papers_from_sqlite(sqlite_path, conference, year)

Import papers for a given conference and year from a SQLite file.

Existing papers for the given conference/year are replaced (not merged). Hierarchical label cache entries that match the conference and year are replaced. Embeddings metadata is validated for consistency (the embedding model must match).

Parameters:

• sqlite_path (Path) – Path to the source SQLite file.

• conference (str) – Conference name being imported.

• year (int) – Year being imported.

Returns:

Number of papers imported.

Return type:

int

Raises:

DatabaseError – If the import fails or the embedding model is inconsistent.

export_clustering_cache_to_json(conference, year)

Export clustering cache entries matching conference and year as JSON.

Parameters:

• conference (str) – Conference name to match.

• year (int) – Year to match.

Returns:

A JSON-serialisable dictionary with an entries list. Each entry contains all ClusteringCache columns except id (auto-generated on import).

Return type:

Dict[str, Any]

Raises:

DatabaseError – If the export fails.

import_clustering_cache_from_json(data, conference, year, overwrite_embedding_model=None)

Import clustering cache entries from a JSON dictionary.

Existing clustering cache entries matching conference and year are deleted before importing the new entries.

Parameters:

• data (Dict[str, Any]) – Dictionary previously returned by export_clustering_cache_to_json().

• conference (str) – Conference name for scoping the delete.

• year (int) – Year for scoping the delete.

• overwrite_embedding_model (Optional[str]) – When provided, the embedding_model field of every imported entry is replaced with this value. Use this when importing an artifact whose embedding model differs from the locally configured one (i.e. when --ignore-embedding-model-mismatch was passed) so that get_clustering_cache() can find the entries using the local model name.

Returns:

Number of cache entries imported.

Return type:

int

Raises:

DatabaseError – If the import fails.

Usage Examples

Basic Operations

DatabaseManager reads its connection settings from the application configuration (PAPER_DB). Papers are added as validated LightweightPaper objects.

from abstracts_explorer.database import DatabaseManager
from abstracts_explorer.plugin import LightweightPaper

db = DatabaseManager()
with db:
    db.create_tables()

    paper = LightweightPaper(
        title="Example Paper",
        authors=["John Doe", "Jane Smith"],
        abstract="This is an example abstract.",
        session="Poster Session 1",
        poster_position="P1",
        year=2025,
        conference="NeurIPS",
        paper_pdf_url="https://example.com/paper.pdf",
    )
    paper_uid = db.add_paper(paper)

Searching Papers

# Keyword search across title, abstract, and keywords
results = db.search_papers(keyword="transformer")

# Field-specific substring search
results = db.search_papers(field_filters={"title": "attention"})
results = db.search_papers(field_filters={"authors": "John Doe"})

# Filter by year and/or conference
papers_2025 = db.search_papers(years=[2025])
neurips_2025 = db.search_papers(conferences=["NeurIPS"], years=[2025])

# Structured filters with get_paper, the single paper-search authority
results = db.get_paper(
    {
        "$and": [
            {"year": 2025},
            {"conference": "NeurIPS"},
        ]
    }
)

Retrieving Data

# Get all papers (limit=0 disables the default limit)
all_papers = db.get_paper(limit=0)

# Get a specific paper by UID
paper = db.get_paper_by_uid(paper_uid)

# Get multiple papers by UID
papers = db.get_papers_by_uids([paper_uid, another_uid])

# Authors are stored on the paper record as a semicolon-separated string
authors = paper["authors"] if paper else None

Database Schema

papers Table

Column	Type	Description
uid	TEXT	Primary key (hash-based unique identifier)
original_id	TEXT	Original ID from the source
title	TEXT	Paper title
authors	TEXT	Semicolon-separated author names
abstract	TEXT	Paper abstract
session	TEXT	Conference session name
poster_position	TEXT	Poster position identifier
paper_pdf_url	TEXT	URL to paper PDF
poster_image_url	TEXT	URL to poster image
url	TEXT	General URL for the paper
room_name	TEXT	Room name for presentation
keywords	TEXT	Comma-separated keywords
starttime	TEXT	Presentation start time
endtime	TEXT	Presentation end time
award	TEXT	Award received (e.g. “Best Paper”)
year	INTEGER	Publication year
conference	TEXT	Conference name (e.g. “NeurIPS”, “ML4PS”)
created_at	TIMESTAMP	Creation timestamp

Additional tables store embeddings metadata, clustering caches, validation data, chat donations, and evaluation data. See abstracts_explorer.db_models for the full model definitions.

Error Handling

Database operations raise DatabaseError:

from abstracts_explorer.database import DatabaseError

try:
    paper_uid = db.add_paper(paper)
except DatabaseError as exc:
    print(f"Database error: {exc}")