Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 29 additions & 7 deletions src/bioregistry/app/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from fastapi.responses import JSONResponse, Response
from pydantic import BaseModel, Field

from .utils import FORMAT_MAP, _autocomplete, _search
from .utils import FORMAT_MAP, AutocompleteResult, PrefixSearchResult, _autocomplete, _search
from ..export.rdf_export import (
collection_to_rdf_str,
metaresource_to_rdf_str,
Expand Down Expand Up @@ -721,19 +721,41 @@ def generate_context_json_ld(
)


@api_router.get("/autocomplete", tags=["search"])
@api_router.get("/autocomplete", tags=["search"], response_model_exclude_none=True)
def autocomplete(
manager: DependsManager,
q: Annotated[str, Query(description="A query for the prefix")],
) -> JSONResponse:
) -> AutocompleteResult:
"""Complete a resolution query."""
return JSONResponse(_autocomplete(manager, q))
return _autocomplete(manager, q)


@api_router.get("/search", tags=["search"])
@api_router.get("/search", tags=["search"], response_model_exclude_none=True)
def search(
manager: DependsManager,
q: Annotated[str, Query(description="A query for the prefix")],
) -> JSONResponse:
) -> list[PrefixSearchResult]:
"""Search for a prefix."""
return JSONResponse(_search(manager, q))
return _search(manager, q)


class ReverseURILookupResults(BaseModel):
"""Results from reverse URI lookup."""

query: str
resources: list[Resource]


@api_router.get("/reverse", tags=["search"])
def reverse_uri_lookup(
manager: DependsManager,
uri_prefix: Annotated[str, Query(description="A URI prefix for lookup")],
) -> ReverseURILookupResults:
"""Lookup a URI, inspired by https://prefix.cc/reverse."""
return ReverseURILookupResults(
query=uri_prefix,
resources=[
manager.rasterized_resource(resource)
for resource in manager.lookup_uri_prefix(uri_prefix)
],
)
119 changes: 79 additions & 40 deletions src/bioregistry/app/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,12 @@
from __future__ import annotations

import json
from collections.abc import Callable, Mapping, Sequence
from collections import defaultdict
from collections.abc import Callable, Sequence
from functools import partial
from typing import Any, cast
from typing import cast

import curies
import werkzeug
import yaml
from flask import (
Expand Down Expand Up @@ -71,22 +73,53 @@ def _normalize_prefix_or_404(
return norm_prefix


def _search(manager_: Manager, q: str) -> list[tuple[str, str]]:
q_norm = _norm(q)
results = [
(prefix, lookup if _norm(prefix) != lookup else "")
for lookup, prefix in manager_.synonyms.items()
if q_norm in lookup
class PrefixLookup(BaseModel):
"""A prefix lookup."""

lookup: str
position: int
complete: bool


class PrefixSearchResult(BaseModel):
"""Results from searching for a prefix."""

prefix: str
lookups: list[PrefixLookup]


def _search(manager_: Manager, query: str) -> list[PrefixSearchResult]:
query = _norm(query)
prefix_to_lookups = defaultdict(list)
for lookup, prefix in manager_.synonyms.items():
if query in lookup:
start = lookup.find(query)
prefix_to_lookups[prefix].append(
PrefixLookup(lookup=lookup, position=start, complete=query == lookup)
)
return [
PrefixSearchResult(prefix=prefix, lookups=lookups)
for prefix, lookups in prefix_to_lookups.items()
]
return sorted(results)


def _autocomplete(manager_: Manager, q: str, url_prefix: str | None = None) -> Mapping[str, Any]:
class AutocompleteResult(BaseModel):
"""Results from autocomplete."""

query: str
success: bool
reason: str
results: curies.Reference | list[PrefixSearchResult]
url: str | None = None
pattern: str | None = None


def _autocomplete(manager_: Manager, q: str, base_url: str | None = None) -> AutocompleteResult:
r"""Run the autocomplete algorithm.

:param manager_: A manager
:param q: The query string
:param url_prefix:
:param base_url:
The explicit URL prefix. If not used, relative paths are generated. Introduced to
solve https://github.com/biopragmatics/bioregistry/issues/596.
:return: A dictionary with the autocomplete results.
Expand All @@ -102,6 +135,9 @@ def _autocomplete(manager_: Manager, q: str, url_prefix: str | None = None) -> M
>>> _autocomplete(manager, "chebi")
{'query': 'chebi', 'results': [('chebi', ''), ('chebi', 'chebiid'), ('goche', 'gochebi')], 'success': True, 'reason': 'matched prefix', 'url': '/chebi'}

>>> _autocomplete(manager, "http://purl.allotrope.org/ontologies/equipment#AFE_")
{'query': 'http://purl.allotrope.org/ontologies/equipment#AFE_', 'results': [('allotrope.equiment', '')], 'success': True, 'reason': 'matched prefix', 'url': '/allotrope.equiment'}

Not matching the pattern:

>>> _autocomplete(manager, "chebi:NOPE")
Expand All @@ -112,59 +148,62 @@ def _autocomplete(manager_: Manager, q: str, url_prefix: str | None = None) -> M
>>> _autocomplete(manager, "chebi:1234")
{'query': 'chebi:1234', 'prefix': 'chebi', 'pattern': '^\\d+$', 'identifier': '1234', 'success': True, 'reason': 'passed validation', 'url': '/chebi:1234'}
"""
if url_prefix is None:
url_prefix = ""
url_prefix = url_prefix.rstrip().rstrip("/")
if base_url is None:
base_url = ""
base_url = base_url.rstrip().rstrip("/")

if q.startswith("https://") or q.startswith("http://"):
raise NotImplementedError

if ":" not in q:
url: str | None
if q in manager_.registry:
reason = "matched prefix"
url = f"{url_prefix}/{q}"
url = f"{base_url}/{q}"
else:
reason = "searched prefix"
url = None
return {
"query": q,
"results": _search(manager_, q),
"success": True,
"reason": reason,
"url": url,
}
return AutocompleteResult(
query=q,
success=True,
reason=reason,
url=url,
results=_search(manager_, q),
)

prefix, identifier = q.split(":", 1)
resource = manager_.get_resource(prefix)
if resource is None:
return {
"query": q,
"prefix": prefix,
"identifier": identifier,
"success": False,
"reason": "bad prefix",
}
return AutocompleteResult(
query=q,
success=False,
reason="bad prefix",
results=curies.Reference(prefix=prefix, identifier=identifier),
)
pattern = manager_.get_pattern(prefix)
if pattern is None:
success = True
reason = "no pattern"
norm_id = resource.standardize_identifier(identifier)
url = f"{url_prefix}/{resource.get_curie(norm_id)}"
url = f"{base_url}/{resource.get_curie(norm_id)}"
elif resource.is_standardizable_identifier(identifier):
success = True
reason = "passed validation"
norm_id = resource.standardize_identifier(identifier)
url = f"{url_prefix}/{resource.get_curie(norm_id)}"
url = f"{base_url}/{resource.get_curie(norm_id)}"
else:
success = False
reason = "failed validation"
url = None
return {
"query": q,
"prefix": prefix,
"pattern": pattern,
"identifier": identifier,
"success": success,
"reason": reason,
"url": url,
}

return AutocompleteResult(
query=q,
success=success,
reason=reason,
url=url,
results=curies.Reference(prefix=prefix, identifier=identifier),
pattern=pattern,
)


def serialize(
Expand Down
20 changes: 20 additions & 0 deletions src/bioregistry/resource_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -2281,6 +2281,26 @@ def get_collection_first_party(
calls[prefix] = False
return calls

def lookup_uri_prefix(self, query: str) -> list[Resource]:
"""Search for resources whose URI prefix begins with the given string.

:param query: A query string to check URI prefixes against
:return: A list of prefixes whose URI prefixes start with
the given URI prefix

>>> from bioregistry import manager
>>> resources = manager.lookup_uri_prefix(
... "http://purl.allotrope.org/ontologies/equipment#AFE_"
... )
>>> resources[0].prefix
'allotrope.equipment
"""
return [
resource
for resource in self.registry.values()
if any(uri_prefix.startswith(query) for uri_prefix in resource.get_uri_prefixes())
]


def _read_contributors(
registry: dict[str, Resource],
Expand Down
1 change: 1 addition & 0 deletions src/bioregistry/validate/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -192,6 +192,7 @@ def _get_all_messages(
def _get_suggestions(uri_prefix: str) -> list[tuple[str, str]] | str:
suggies = []
for x, y in rpm.items():
# TODO incorporate more principled URI prefix search here
if x.startswith(uri_prefix):
suggies.append((x, y))
if x == uri_prefix:
Expand Down
18 changes: 18 additions & 0 deletions tests/test_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -285,3 +285,21 @@ def test_add_resource(self) -> None:

self.assertIsNotNone(manager.get_resource(test_prefix))
self.assertIsNotNone(manager.get_resource(test_synonym))

def test_search_uri_prefix(self) -> None:
"""Test searching URI prefixes."""
results = self.manager.lookup_uri_prefix(
"http://purl.allotrope.org/ontologies/equipment#AFE_"
)
self.assertEqual(1, len(results))
self.assertEqual("allotrope.equipment", results[0].prefix)

results = self.manager.lookup_uri_prefix(
"http://purl.obolibrary.org/obo",
)
self.assertLess(200, len(results))
prefixes = {r.prefix for r in results}
# check the top-level one makes it
self.assertIn("obo", prefixes)
# check a sub-URI-prefix makes it
self.assertIn("go", prefixes)
Loading