annif.rest.suggest() - Code Metrics - Inspection of "Support for batch suggest operations in suggest an..." - NatLibFi/Annif - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#663)

by Juho

created 2023-01-30 08:10 UTC

annif.rest.suggest() A

↳ Parent: annif.rest

Complexity

Conditions

Size

Total Lines	14
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	8
nop	2
dl	0
loc	14
rs	10
c	0
b	0
f	0

"""Definitions for REST API operations. These are wired via Connexion to
methods defined in the Swagger specification."""

import importlib

import connexion

import annif.registry
from annif.corpus import Document, DocumentList, SubjectSet
from annif.exception import AnnifException
from annif.project import Access
from annif.suggestion import SuggestionFilter


def project_not_found_error(project_id):
    """return a Connexion error object when a project is not found"""

    return connexion.problem(
        status=404,
        title="Project not found",
        detail="Project '{}' not found".format(project_id),
    )


def server_error(err):
    """return a Connexion error object when there is a server error (project
    or backend problem)"""

    return connexion.problem(
        status=503, title="Service unavailable", detail=err.format_message()
    )


def language_not_supported_error(lang):
    """return a Connexion error object when attempting to use unsupported language"""

    return connexion.problem(
        status=400,
        title="Bad Request",
        detail=f'language "{lang}" not supported by vocabulary',
    )


def show_info():
    """return version of annif and a title for the api according to Swagger spec"""

    return {"title": "Annif REST API", "version": importlib.metadata.version("annif")}


def list_projects():
    """return a dict with projects formatted according to Swagger spec"""

    return {
        "projects": [
            proj.dump()
            for proj in annif.registry.get_projects(min_access=Access.public).values()
        ]
    }


def show_project(project_id):
    """return a single project formatted according to Swagger spec"""

    try:
        project = annif.registry.get_project(project_id, min_access=Access.hidden)
    except ValueError:
        return project_not_found_error(project_id)
    return project.dump()


def _suggestion_to_dict(suggestion, subject_index, language):
    subject = subject_index[suggestion.subject_id]
    return {
        "uri": subject.uri,
        "label": subject.labels[language],
        "notation": subject.notation,
        "score": suggestion.score,
    }


def _hit_sets_to_list(hit_sets, hit_filter, subjects, lang):
    return [
        {
            "results": [
                _suggestion_to_dict(hit, subjects, lang)
                for hit in hit_filter(hits).as_list()
            ]
        }
        for hits in hit_sets
    ]


def suggest(project_id, body):
    """suggest subjects for the given text and return a dict with results
    formatted according to Swagger spec"""

    parameters = dict(
        (key, body[key]) for key in ["language", "limit", "threshold"] if key in body
    )
    documents = [{"text": body["text"]}]
    result = _suggest(project_id, documents, parameters)

    if isinstance(result, list):
        return result[0]  # successful operation
    else:
        return result  # connexion problem


def suggest_batch(project_id, body):
    """suggest subjects for the given documents and return a list of dicts with results
    formatted according to Swagger spec"""

    parameters = body.get("parameters", {})
    documents = body["documents"]
    result = _suggest(project_id, documents, parameters)

    if isinstance(result, list):
        for document_results, document in zip(result, documents):
            document_results["id"] = document.get("id")
    return result


def _suggest(project_id, documents, parameters):
    corpus = _documents_to_corpus(documents, subject_index=None)
    try:
        project = annif.registry.get_project(project_id, min_access=Access.hidden)
    except ValueError:
        return project_not_found_error(project_id)

    try:
        lang = parameters.get("language") or project.vocab_lang
    except AnnifException as err:
        return server_error(err)

    if lang not in project.vocab.languages:
        return language_not_supported_error(lang)

    limit = parameters.get("limit", 10)
    threshold = parameters.get("threshold", 0.0)

    try:
        hit_filter = SuggestionFilter(project.subjects, limit, threshold)
        hit_sets = project.suggest_batch(corpus)
    except AnnifException as err:
        return server_error(err)

    return _hit_sets_to_list(hit_sets, hit_filter, project.subjects, lang)


def _documents_to_corpus(documents, subject_index):
    if subject_index is not None:
        corpus = [
            Document(
                text=d["text"],
                subject_set=SubjectSet(
                    [subject_index.by_uri(subj["uri"]) for subj in d["subjects"]]
                ),
            )
            for d in documents
            if "text" in d and "subjects" in d
        ]
    else:
        corpus = [
            Document(text=d["text"], subject_set=None) for d in documents if "text" in d
        ]
    return DocumentList(corpus)


def learn(project_id, body):
    """learn from documents and return an empty 204 response if succesful"""

    try:
        project = annif.registry.get_project(project_id, min_access=Access.hidden)
    except ValueError:
        return project_not_found_error(project_id)

    try:
        corpus = _documents_to_corpus(body, project.subjects)
        project.learn(corpus)
    except AnnifException as err:
        return server_error(err)

    return None, 204


1			"""Definitions for REST API operations. These are wired via Connexion to
2			methods defined in the Swagger specification."""
3
4			import importlib
5
6			import connexion
7
8			import annif.registry
9			from annif.corpus import Document, DocumentList, SubjectSet
10			from annif.exception import AnnifException
11			from annif.project import Access
12			from annif.suggestion import SuggestionFilter
13
14
15			def project_not_found_error(project_id):
16			"""return a Connexion error object when a project is not found"""
17
18			return connexion.problem(
19			status=404,
20			title="Project not found",
21			detail="Project '{}' not found".format(project_id),
22			)
23
24
25			def server_error(err):
26			"""return a Connexion error object when there is a server error (project
27			or backend problem)"""
28
29			return connexion.problem(
30			status=503, title="Service unavailable", detail=err.format_message()
31			)
32
33
34			def language_not_supported_error(lang):
35			"""return a Connexion error object when attempting to use unsupported language"""
36
37			return connexion.problem(
38			status=400,
39			title="Bad Request",
40			detail=f'language "{lang}" not supported by vocabulary',
41			)
42
43
44			def show_info():
45			"""return version of annif and a title for the api according to Swagger spec"""
46
47			return {"title": "Annif REST API", "version": importlib.metadata.version("annif")}
48
49
50			def list_projects():
51			"""return a dict with projects formatted according to Swagger spec"""
52
53			return {
54			"projects": [
55			proj.dump()
56			for proj in annif.registry.get_projects(min_access=Access.public).values()
57			]
58			}
59
60
61			def show_project(project_id):
62			"""return a single project formatted according to Swagger spec"""
63
64			try:
65			project = annif.registry.get_project(project_id, min_access=Access.hidden)
66			except ValueError:
67			return project_not_found_error(project_id)
68			return project.dump()
69
70
71			def _suggestion_to_dict(suggestion, subject_index, language):
72			subject = subject_index[suggestion.subject_id]
73			return {
74			"uri": subject.uri,
75			"label": subject.labels[language],
76			"notation": subject.notation,
77			"score": suggestion.score,
78			}
79
80
81			def _hit_sets_to_list(hit_sets, hit_filter, subjects, lang):
82			return [
83			{
84			"results": [
85			_suggestion_to_dict(hit, subjects, lang)
86			for hit in hit_filter(hits).as_list()
87			]
88			}
89			for hits in hit_sets
90			]
91
92
93			def suggest(project_id, body):
94			"""suggest subjects for the given text and return a dict with results
95			formatted according to Swagger spec"""
96
97			parameters = dict(
98			(key, body[key]) for key in ["language", "limit", "threshold"] if key in body
99			)
100			documents = [{"text": body["text"]}]
101			result = _suggest(project_id, documents, parameters)
102
103			if isinstance(result, list):
104			return result[0] # successful operation
105			else:
106			return result # connexion problem
107
108
109			def suggest_batch(project_id, body):
110			"""suggest subjects for the given documents and return a list of dicts with results
111			formatted according to Swagger spec"""
112
113			parameters = body.get("parameters", {})
114			documents = body["documents"]
115			result = _suggest(project_id, documents, parameters)
116
117			if isinstance(result, list):
118			for document_results, document in zip(result, documents):
119			document_results["id"] = document.get("id")
120			return result
121
122
123			def _suggest(project_id, documents, parameters):
124			corpus = _documents_to_corpus(documents, subject_index=None)
125			try:
126			project = annif.registry.get_project(project_id, min_access=Access.hidden)
127			except ValueError:
128			return project_not_found_error(project_id)
129
130			try:
131			lang = parameters.get("language") or project.vocab_lang
132			except AnnifException as err:
133			return server_error(err)
134
135			if lang not in project.vocab.languages:
136			return language_not_supported_error(lang)
137
138			limit = parameters.get("limit", 10)
139			threshold = parameters.get("threshold", 0.0)
140
141			try:
142			hit_filter = SuggestionFilter(project.subjects, limit, threshold)
143			hit_sets = project.suggest_batch(corpus)
144			except AnnifException as err:
145			return server_error(err)
146
147			return _hit_sets_to_list(hit_sets, hit_filter, project.subjects, lang)
148
149
150			def _documents_to_corpus(documents, subject_index):
151			if subject_index is not None:
152			corpus = [
153			Document(
154			text=d["text"],
155			subject_set=SubjectSet(
156			[subject_index.by_uri(subj["uri"]) for subj in d["subjects"]]
157			),
158			)
159			for d in documents
160			if "text" in d and "subjects" in d
161			]
162			else:
163			corpus = [
164			Document(text=d["text"], subject_set=None) for d in documents if "text" in d
165			]
166			return DocumentList(corpus)
167
168
169			def learn(project_id, body):
170			"""learn from documents and return an empty 204 response if succesful"""
171
172			try:
173			project = annif.registry.get_project(project_id, min_access=Access.hidden)
174			except ValueError:
175			return project_not_found_error(project_id)
176
177			try:
178			corpus = _documents_to_corpus(body, project.subjects)
179			project.learn(corpus)
180			except AnnifException as err:
181			return server_error(err)
182
183			return None, 204
184

NatLibFi / Annif

Pull Request — master (#663)

annif.rest.suggest() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like