typing added

Author: bodintsov

mypy.ini

@@ -0,0 +1,22 @@
+[mypy]
+python_version = 3.13
+plugins = mypy_django_plugin.main
+
+strict = True
+ignore_missing_imports = True
+show_column_numbers = True
+pretty = True
+
+# Вимикаємо перевірку всього, що не готове
+exclude = ^(benchmarks/|github/|api/|app/|db/|how-to/|osf_ouath2_adapter/|project/|share/|share.egg-info/|templates/|tests/|venv/)
+
+disallow_untyped_defs = True
+warn_unused_ignores = True
+warn_redundant_casts = True
+warn_return_any = True
+
+[mypy.plugins.django-stubs]
+django_settings_module = project.settings
+
+[mypy-trove.derive.osfmap_json]
+disable_error_code = no-any-return

share/admin/util.py

@@ -1,12 +1,15 @@
+from typing import Callable, Any, Optional, Sequence, Type, TypeVar
+
 from django.contrib.admin import SimpleListFilter
 from django.core.paginator import Paginator
 from django.db import connection, transaction, OperationalError
+from django.db.models import Model
 from django.utils.functional import cached_property
 from django.urls import reverse
 from django.utils.html import format_html
 
 from share.models import SourceConfig
-
+T = TypeVar('T', bound=Type[Model])
 
 # TimeLimitedPaginator from https://hakibenita.com/optimizing-the-django-admin-paginator
 class TimeLimitedPaginator(Paginator):
@@ -46,7 +49,7 @@ def admin_link_html(linked_obj):
     return format_html('<a href="{}">{}</a>', url, repr(linked_obj))
 
 
-def linked_fk(field_name):
+def linked_fk(field_name: str) -> Callable[[Any], Any]:
     """Decorator that adds a link for a foreign key field
     """
     def add_link(cls):
@@ -62,11 +65,15 @@ def link(self, instance):
     return add_link
 
 
-def linked_many(field_name, order_by=None, select_related=None, defer=None):
-    """Decorator that adds links for a *-to-many field
-    """
-    def add_links(cls):
-        def links(self, instance):
+def linked_many(
+    field_name: str,
+    order_by: Optional[Sequence[str]] = None,
+    select_related: Optional[Sequence[str]] = None,
+    defer: Optional[Sequence[str]] = None,
+) -> Callable[[T], T]:
+    """Decorator that adds links for a *-to-many field"""
+    def add_links(cls: T) -> T:
+        def links(self, instance: Model) -> str:
             linked_qs = getattr(instance, field_name).all()
             if select_related:
                 linked_qs = linked_qs.select_related(*select_related)
@@ -81,15 +88,14 @@ def links(self, instance):
                     for obj in linked_qs
                 ))
             )
-        links_field = '{}_links'.format(field_name)
+        links_field = f'{field_name}_links'
         links.short_description = field_name.replace('_', ' ')
         setattr(cls, links_field, links)
         append_to_cls_property(cls, 'readonly_fields', links_field)
         append_to_cls_property(cls, 'exclude', field_name)
         return cls
     return add_links
 
-
 class SourceConfigFilter(SimpleListFilter):
     title = 'Source Config'
     parameter_name = 'source_config'

share/models/ingest.py

@@ -70,7 +70,7 @@ def __str__(self):
 
 
 class SourceConfigManager(NaturalKeyManager):
-    def get_or_create_push_config(self, user, transformer_key=None):
+    def get_or_create_push_config(self, user, transformer_key=None) -> 'SourceConfig':
         assert isinstance(user, ShareUser)
         _config_label = '.'.join((
             user.username,
@@ -231,7 +231,7 @@ class RawDatum(models.Model):
 
     objects = RawDatumManager()
 
-    def is_latest(self):
+    def is_latest(self) -> bool:
         return (
             RawDatum.objects
             .latest_by_suid_id(self.suid_id)

share/oaipmh/util.py

@@ -1,11 +1,15 @@
+import datetime
+from typing import Type, Optional
+from xml.etree.ElementTree import SubElement
+
 from lxml import etree
 from primitive_metadata import primitive_rdf
 
 from share.util.fromisoformat import fromisoformat
 from trove.vocab.namespaces import OAI, OAI_DC
 
 
-def format_datetime(dt):
+def format_datetime(dt) -> datetime:
     """OAI-PMH has specific time format requirements -- comply.
     """
     if isinstance(dt, primitive_rdf.Literal):
@@ -25,15 +29,15 @@ def format_datetime(dt):
 }
 
 
-def ns(namespace_prefix, tag_name):
+def ns(namespace_prefix, tag_name) -> str:
     """format XML tag/attribute name with full namespace URI
 
     see https://lxml.de/tutorial.html#namespaces
     """
     return f'{{{XML_NAMESPACES[namespace_prefix]}}}{tag_name}'
 
 
-def nsmap(*namespace_prefixes, default=None):
+def nsmap(*namespace_prefixes, default=None) -> dict[Optional[str], str]:
     """build a namespace map suitable for lxml
 
     see https://lxml.de/tutorial.html#namespaces
@@ -49,7 +53,7 @@ def nsmap(*namespace_prefixes, default=None):
 
 
 # wrapper for lxml.etree.SubElement, adds `text` kwarg for convenience
-def SubEl(parent, tag_name, text=None, **kwargs):
+def SubEl(parent, tag_name, text=None, **kwargs) -> Type[SubElement]:
     element = etree.SubElement(parent, tag_name, **kwargs)
     if isinstance(text, primitive_rdf.Literal):
         _language_tag = text.language

share/search/index_messenger.py

@@ -12,7 +12,7 @@
 
 from share.search.messages import MessagesChunk, MessageType
 from share.search import index_strategy
-
+from trove.models import Indexcard
 
 logger = logging.getLogger(__name__)
 
@@ -25,15 +25,15 @@ class IndexMessenger:
         'max_retries': 30,    # give up after 30 tries.
     }
 
-    def __init__(self, *, celery_app=None, index_strategys=None):
+    def __init__(self, *, celery_app=None, index_strategys=None) -> None:
         self.celery_app = (
             celery.current_app
             if celery_app is None
             else celery_app
         )
         self.index_strategys = index_strategys or tuple(index_strategy.each_strategy())
 
-    def notify_indexcard_update(self, indexcards, *, urgent=False):
+    def notify_indexcard_update(self, indexcards: list[Indexcard], *, urgent=False) -> None:
         self.send_messages_chunk(
             MessagesChunk(
                 MessageType.UPDATE_INDEXCARD,
@@ -53,7 +53,7 @@ def notify_indexcard_update(self, indexcards, *, urgent=False):
             urgent=urgent,
         )
 
-    def notify_suid_update(self, suid_ids, *, urgent=False):
+    def notify_suid_update(self, suid_ids, *, urgent=False) -> None:
         self.send_messages_chunk(
             MessagesChunk(MessageType.INDEX_SUID, suid_ids),
             urgent=urgent,

share/util/__init__.py

@@ -24,7 +24,7 @@ class IDObfuscator:
     ID_RE = re.compile(r'([0-9A-Fa-f]{2,})([0-9A-Fa-f]{3})-([0-9A-Fa-f]{3})-([0-9A-Fa-f]{3})')
 
     @classmethod
-    def encode(cls, instance):
+    def encode(cls, instance) -> str:
         return cls.encode_id(instance.id, instance._meta.model)
 
     @classmethod

share/util/checksum_iri.py

@@ -1,6 +1,7 @@
 import dataclasses
 import hashlib
 import json
+from typing import Self
 
 
 def _ensure_bytes(bytes_or_something) -> bytes:
@@ -37,7 +38,7 @@ def __str__(self):
         return f'urn:checksum:{self.checksumalgorithm_name}:{self.salt}:{self.hexdigest}'
 
     @classmethod
-    def digest(cls, checksumalgorithm_name, *, salt, raw_data):
+    def digest(cls, checksumalgorithm_name, *, salt, raw_data) -> Self:
         try:
             hexdigest_fn = CHECKSUM_ALGORITHMS[checksumalgorithm_name]
         except KeyError:

trove/admin.py

@@ -1,3 +1,4 @@
+from typing import Any, Self
 from django.contrib import admin
 from django.utils.html import format_html
 
@@ -17,7 +18,7 @@
 @admin.register(ResourceIdentifier, site=admin_site)
 @linked_many('suid_set')
 @linked_many('indexcard_set')
-class ResourceIdentifierAdmin(admin.ModelAdmin):
+class ResourceIdentifierAdmin(admin.ModelAdmin): #type: ignore
     readonly_fields = (
         'created',
         'modified',
@@ -37,7 +38,7 @@ class ResourceIdentifierAdmin(admin.ModelAdmin):
 @linked_fk('source_record_suid')
 @linked_many('focustype_identifier_set')
 @linked_many('focus_identifier_set')
-class IndexcardAdmin(admin.ModelAdmin):
+class IndexcardAdmin(admin.ModelAdmin): #type: ignore
     readonly_fields = (
         'uuid',
         'created',
@@ -52,15 +53,15 @@ class IndexcardAdmin(admin.ModelAdmin):
     list_filter = ('deleted', 'source_record_suid__source_config')
     actions = ('_freshen_index',)
 
-    def _freshen_index(self, request, queryset):
+    def _freshen_index(self, queryset: list[Indexcard]) -> None:
         IndexMessenger().notify_indexcard_update(queryset)
-    _freshen_index.short_description = 'freshen indexcard in search index'
+    _freshen_index.short_description = 'freshen indexcard in search index' # type: ignore
 
 
 @admin.register(LatestIndexcardRdf, site=admin_site)
 @linked_fk('from_raw_datum')
 @linked_fk('indexcard')
-class LatestIndexcardRdfAdmin(admin.ModelAdmin):
+class LatestIndexcardRdfAdmin(admin.ModelAdmin): #type: ignore
     readonly_fields = (
         'created',
         'modified',
@@ -74,15 +75,15 @@ class LatestIndexcardRdfAdmin(admin.ModelAdmin):
     list_select_related = ('indexcard',)
     show_full_result_count = False
 
-    def rdf_as_turtle__pre(self, instance):
+    def rdf_as_turtle__pre(self, instance: Any) -> str:
         return format_html('<pre>{}</pre>', instance.rdf_as_turtle)
-    rdf_as_turtle__pre.short_description = 'rdf as turtle'
+    rdf_as_turtle__pre.short_description = 'rdf as turtle' # type: ignore[attr-defined]
 
 
 @admin.register(ArchivedIndexcardRdf, site=admin_site)
 @linked_fk('from_raw_datum')
 @linked_fk('indexcard')
-class ArchivedIndexcardRdfAdmin(admin.ModelAdmin):
+class ArchivedIndexcardRdfAdmin(admin.ModelAdmin): #type: ignore
     readonly_fields = (
         'created',
         'modified',
@@ -96,16 +97,16 @@ class ArchivedIndexcardRdfAdmin(admin.ModelAdmin):
     list_select_related = ('indexcard', 'from_raw_datum',)
     show_full_result_count = False
 
-    def rdf_as_turtle__pre(self, instance):
+    def rdf_as_turtle__pre(self, instance: Any) -> str:
         return format_html('<pre>{}</pre>', instance.rdf_as_turtle)
-    rdf_as_turtle__pre.short_description = 'rdf as turtle'
+    rdf_as_turtle__pre.short_description = 'rdf as turtle' # type: ignore[attr-defined]
 
 
 @admin.register(SupplementaryIndexcardRdf, site=admin_site)
 @linked_fk('from_raw_datum')
 @linked_fk('indexcard')
 @linked_fk('supplementary_suid')
-class SupplementaryIndexcardRdfAdmin(admin.ModelAdmin):
+class SupplementaryIndexcardRdfAdmin(admin.ModelAdmin): #type: ignore
     readonly_fields = (
         'created',
         'modified',
@@ -119,15 +120,14 @@ class SupplementaryIndexcardRdfAdmin(admin.ModelAdmin):
     list_select_related = ('indexcard', 'from_raw_datum',)
     show_full_result_count = False
 
-    def rdf_as_turtle__pre(self, instance):
+    def rdf_as_turtle__pre(self, instance: SupplementaryIndexcardRdf) -> str:
         return format_html('<pre>{}</pre>', instance.rdf_as_turtle)
-    rdf_as_turtle__pre.short_description = 'rdf as turtle'
-
+    rdf_as_turtle__pre.short_description = 'rdf as turtle' # type: ignore[attr-defined]
 
 @admin.register(DerivedIndexcard, site=admin_site)
 @linked_fk('upriver_indexcard')
 @linked_fk('deriver_identifier')
-class DerivedIndexcardAdmin(admin.ModelAdmin):
+class DerivedIndexcardAdmin(admin.ModelAdmin): #type: ignore
     readonly_fields = (
         'created',
         'modified',

trove/derive/__init__.py

@@ -1,3 +1,4 @@
+from typing import Iterable, Type, Union
 from . import (
     sharev2_elastic,
     osfmap_json,
@@ -16,8 +17,15 @@
     # osfmap_jsonld_minimal?
 )
 
+DeriverType = Type[
+    sharev2_elastic.ShareV2ElasticDeriver |
+    osfmap_json.OsfmapJsonDeriver |
+    oaidc_xml.OaiDcXmlDeriver
+]
 
-def get_deriver_classes(deriver_iri_filter=None):
+def get_deriver_classes(
+    deriver_iri_filter: Iterable[str] | None = None,
+) -> Union[list[DeriverType], tuple[DeriverType, ...]]:
     if deriver_iri_filter is None:
         return DERIVER_SET
     return [

trove/derive/_base.py

@@ -1,4 +1,5 @@
 import abc
+from typing import Any
 
 from primitive_metadata import primitive_rdf
 
@@ -15,7 +16,7 @@ def __init__(self, upriver_rdf: IndexcardRdf):
         self.focus_iri = upriver_rdf.focus_iri
         self.data = upriver_rdf.as_rdfdoc_with_supplements()
 
-    def q(self, pathset):
+    def q(self, pathset: Any) -> Any:
         # convenience for querying self.data on self.focus_iri
         return self.data.q(self.focus_iri, pathset)