ruff

ckanext/relationship/helpers.py

@@ -18,7 +18,9 @@ def get_helpers():
     return {f.__name__: f for f in helper_functions}
 
 
-def relationship_get_entity_list(entity, entity_type, include_private=True):
+def relationship_get_entity_list(
+    entity: str, entity_type: str, include_private: bool = True
+) -> list[dict[str, str]]:
     """Return ids list of specified entity (entity, entity_type)."""
     context = {}
     if entity == "package":
@@ -42,7 +44,9 @@ def relationship_get_entity_list(entity, entity_type, include_private=True):
     return entity_list
 
 
-def relationship_get_current_relations_list(data, field) -> list[str]:
+def relationship_get_current_relations_list(
+    data: dict[str, Any], field: dict[str, Any]
+) -> list[str]:
     """Pull existing relations for form_snippet and display_snippet."""
     subject_id = field.get("id")
     subject_name = field.get("name")

ckanext/relationship/logic/action.py

@@ -3,11 +3,13 @@
 from typing import Any
 
 from flask import jsonify
+from flask.wrappers import Response
 from sqlalchemy import or_
 
 import ckan.plugins.toolkit as tk
 from ckan import authz, logic
 from ckan.logic import validate
+from ckan.types import Action, Context
 
 from ckanext.relationship import utils
 from ckanext.relationship.config import views_without_relationships_in_package_show
@@ -31,7 +33,9 @@ def get_actions():
 
 
 @validate(schema.relation_create)
-def relationship_relation_create(context, data_dict) -> list[dict[str, str]]:
+def relationship_relation_create(
+    context: Context, data_dict: dict[str, Any]
+) -> list[dict[str, str]]:
     """Create relation with specified type (relation_type) between two entities
     specified by ids (subject_id, object_id). Also create reverse relation.
     """
@@ -42,7 +46,7 @@ def relationship_relation_create(context, data_dict) -> list[dict[str, str]]:
     relation_type = data_dict.get("relation_type")
 
     if Relationship.by_object_id(subject_id, object_id, relation_type):
-        return None
+        return []
 
     relation = Relationship(
         subject_id=subject_id,
@@ -64,7 +68,9 @@ def relationship_relation_create(context, data_dict) -> list[dict[str, str]]:
 
 
 @validate(schema.relation_delete)
-def relationship_relation_delete(context, data_dict) -> list[dict[str, str]]:
+def relationship_relation_delete(
+    context: Context, data_dict: dict[str, Any]
+) -> list[dict[str, str]]:
     """Delete relation with specified type (relation_type) between two entities
     specified by ids (subject_id, object_id). Also delete reverse relation.
     """
@@ -126,7 +132,9 @@ def relationship_relation_delete(context, data_dict) -> list[dict[str, str]]:
 
 
 @validate(schema.relations_list)
-def relationship_relations_list(context, data_dict) -> list[dict[str, str]]:
+def relationship_relations_list(
+    context: Context, data_dict: dict[str, Any]
+) -> list[dict[str, str]]:
     """Return a list of dictionaries representing the relations of a specified entity
     (object_entity, object_type) related to the specified type of relation
     (relation_type) with an entity specified by its id (subject_id).
@@ -153,7 +161,9 @@ def relationship_relations_list(context, data_dict) -> list[dict[str, str]]:
 
 
 @validate(schema.relations_ids_list)
-def relationship_relations_ids_list(context, data_dict) -> list[str]:
+def relationship_relations_ids_list(
+    context: Context, data_dict: dict[str, Any]
+) -> list[str]:
     """Return ids list of specified entity (object_entity, object_type) related
     with specified type of relation (relation_type) with entity specified
     by id (subject_id).
@@ -166,7 +176,9 @@ def relationship_relations_ids_list(context, data_dict) -> list[str]:
 
 
 @validate(schema.get_entity_list)
-def relationship_get_entity_list(context, data_dict) -> list[str]:
+def relationship_get_entity_list(
+    context: Context, data_dict: dict[str, Any]
+) -> list[str]:
     """Return ids list of specified entity (entity, entity_type)."""
     tk.check_access("relationship_get_entity_list", context, data_dict)
 
@@ -186,7 +198,7 @@ def relationship_get_entity_list(context, data_dict) -> list[str]:
 
 
 @validate(schema.autocomplete)
-def relationship_autocomplete(context, data_dict) -> dict[str, Any]:
+def relationship_autocomplete(context: Context, data_dict: dict[str, Any]) -> Response:
     fq = f'type:{data_dict["entity_type"]} -id:{data_dict["current_entity_id"]}'
 
     if data_dict.get("owned_only") and not (
@@ -224,7 +236,9 @@ def relationship_autocomplete(context, data_dict) -> dict[str, Any]:
 
 @tk.chained_action
 @tk.side_effect_free
-def package_show(next_, context, data_dict) -> dict[str, Any]:
+def package_show(
+    next_: Action, context: Context, data_dict: dict[str, Any]
+) -> dict[str, Any]:
     result = next_(context, data_dict)
 
     pkg_id = result["id"]

ckanext/relationship/logic/schema.py

@@ -1,8 +1,9 @@
 from ckan.logic.schema import validator_args
+from ckan.types import Schema, Validator
 
 
 @validator_args
-def relation_create(not_empty, one_of):
+def relation_create(not_empty: Validator, one_of: Validator) -> Schema:
     return {
         "subject_id": [
             not_empty,
@@ -17,7 +18,9 @@ def relation_create(not_empty, one_of):
 
 
 @validator_args
-def relation_delete(not_empty, one_of, ignore_missing):
+def relation_delete(
+    not_empty: Validator, one_of: Validator, ignore_missing: Validator
+) -> Schema:
     return {
         "subject_id": [
             not_empty,
@@ -33,7 +36,9 @@ def relation_delete(not_empty, one_of, ignore_missing):
 
 
 @validator_args
-def relations_list(not_empty, one_of, ignore_missing):
+def relations_list(
+    not_empty: Validator, one_of: Validator, ignore_missing: Validator
+) -> Schema:
     return {
         "subject_id": [
             not_empty,
@@ -53,7 +58,9 @@ def relations_list(not_empty, one_of, ignore_missing):
 
 
 @validator_args
-def relations_ids_list(not_empty, one_of, ignore_missing):
+def relations_ids_list(
+    not_empty: Validator, one_of: Validator, ignore_missing: Validator
+) -> Schema:
     return {
         "subject_id": [
             not_empty,
@@ -73,7 +80,7 @@ def relations_ids_list(not_empty, one_of, ignore_missing):
 
 
 @validator_args
-def get_entity_list(not_empty, one_of):
+def get_entity_list(not_empty: Validator, one_of: Validator) -> Schema:
     return {
         "entity": [
             not_empty,
@@ -86,7 +93,7 @@ def get_entity_list(not_empty, one_of):
 
 
 @validator_args
-def autocomplete(not_empty):
+def autocomplete(not_empty: Validator) -> Schema:
     return {
         "incomplete": [],
         "current_entity_id": [

ckanext/relationship/logic/validators.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
 import json
+from typing import Any
 
-from ckantoolkit import missing
 from six import string_types
 
 import ckan.plugins.toolkit as tk
+from ckan.types import Context, FlattenDataDict, FlattenErrorDict, FlattenKey
 
 from ckanext.scheming.validation import (
     scheming_multiple_choice_output,
@@ -20,13 +21,18 @@ def get_validators():
 
 
 @scheming_validator
-def relationship_related_entity(field, schema):
+def relationship_related_entity(field: dict[str, Any], schema: dict[str, Any]):
     related_entity = field.get("related_entity")
     related_entity_type = field.get("related_entity_type")
     relation_type = field.get("relation_type")
 
-    def validator(key, data, errors, context):
-        if field.get("required") and data[key] is missing:
+    def validator(
+        key: FlattenKey,
+        data: FlattenDataDict,
+        errors: FlattenErrorDict,
+        context: Context,
+    ):
+        if field.get("required") and data[key] is tk.missing:
             errors[key].append(tk._("Select at least one"))
 
         entity_id = data.get(("id",))
@@ -54,10 +60,10 @@ def validator(key, data, errors, context):
 
 
 def get_current_relations(
-    entity_id,
-    related_entity,
-    related_entity_type,
-    relation_type,
+    entity_id: str | None,
+    related_entity: str,
+    related_entity_type: str,
+    relation_type: str,
 ):
     if entity_id:
         current_relations = tk.get_action("relationship_relations_list")(
@@ -86,7 +92,7 @@ def get_selected_relations(selected_relations: list | str | None) -> set[str]:
     ):
         selected_relations = selected_relations[0].split(",")
 
-    if selected_relations is not missing:
+    if selected_relations is not tk.missing:
         selected_relations = scheming_multiple_choice_output(selected_relations)
         selected_relations = [] if selected_relations == [""] else selected_relations
     else:

ckanext/relationship/model/relationship.py

@@ -42,7 +42,7 @@ def as_dict(self):
         }
 
     @classmethod
-    def by_object_id(cls, subject_id, object_id, relation_type):
+    def by_object_id(cls, subject_id: str, object_id: str, relation_type: str):
         subject_name = _entity_name_by_id(subject_id)
         object_name = _entity_name_by_id(object_id)
 
@@ -61,11 +61,7 @@ def by_object_id(cls, subject_id, object_id, relation_type):
 
     @classmethod
     def by_subject_id(
-        cls,
-        subject_id,
-        object_entity,
-        object_type=None,
-        relation_type=None,
+        cls, subject_id: str, object_entity: str, object_type: str, relation_type: str
     ):
         subject_name = _entity_name_by_id(subject_id)
 
@@ -91,7 +87,7 @@ def by_subject_id(
         return q.distinct().all()
 
 
-def _entity_name_by_id(entity_id):
+def _entity_name_by_id(entity_id: str):
     """Returns entity (package or organization or group) name by its id."""
     pkg = (
         model.Session.query(model.Package)