OpenConceptLab/ocl_issues#1732 | URL registry lookup operation

core/common/swagger_parameters.py

@@ -64,6 +64,8 @@
 canonical_url_param = openapi.Parameter(
     CANONICAL_URL_REQUEST_PARAM, openapi.IN_QUERY, type=openapi.TYPE_STRING,
 )
+url_registry_url_lookup_param = openapi.Parameter(
+    'url', openapi.IN_QUERY, description="repo url", type=openapi.TYPE_STRING, required=True)
 
 released_param = openapi.Parameter(
     RELEASED_PARAM, openapi.IN_QUERY, type=openapi.TYPE_BOOLEAN, default=False,

core/url_registry/migrations/0003_alter_urlregistry_organization_and_more.py

@@ -0,0 +1,27 @@
+# Generated by Django 4.2.4 on 2024-01-18 10:36
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('orgs', '0021_organization_checksums'),
+        ('url_registry', '0002_remove_urlregistry_global_url_unique_and_more'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='urlregistry',
+            name='organization',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='url_registry_entries', to='orgs.organization'),
+        ),
+        migrations.AlterField(
+            model_name='urlregistry',
+            name='user',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='url_registry_entries', to=settings.AUTH_USER_MODEL),
+        ),
+    ]

core/url_registry/models.py

@@ -1,7 +1,7 @@
 from django.db import models
 from pydash import get
 
-from core.common.models import BaseModel
+from core.common.models import BaseModel, ConceptContainerModel
 
 
 class URLRegistry(BaseModel):
@@ -10,11 +10,11 @@ class URLRegistry(BaseModel):
     namespace = models.CharField(max_length=300, null=True, blank=True)
     organization = models.ForeignKey(
         'orgs.Organization', on_delete=models.CASCADE, null=True, blank=True,
-        related_name='url_registries'
+        related_name='url_registry_entries'
     )
     user = models.ForeignKey(
         'users.UserProfile', on_delete=models.CASCADE, null=True, blank=True,
-        related_name='url_registries'
+        related_name='url_registry_entries'
     )
     public_access = None
     uri = None
@@ -65,17 +65,33 @@ def owner_url(self):
     def owner_type(self):
         return get(self.owner, 'resource_type') or None
 
+    @property
+    def active_entries(self):
+        return URLRegistry.get_active_entries(self.owner)
+
     def is_uniq(self):
-        return not self.get_active_entries().filter(url=self.url).exists()
+        return not self.active_entries.filter(url=self.url).exists()
+
+    @classmethod
+    def get_active_entries(cls, owner):
+        queryset = owner.url_registry_entries if owner else cls.get_global_entries()
+        return queryset.filter(is_active=True)
+
+    @classmethod
+    def get_global_entries(cls):
+        return cls.objects.filter(organization__isnull=True, user__isnull=True)
 
-    def get_active_entries(self):
-        queryset = URLRegistry.objects.filter(is_active=True)
+    @classmethod
+    def get_active_global_entries(cls):
+        return cls.get_global_entries().filter(is_active=True)
 
-        if self.organization:
-            queryset = queryset.filter(organization=self.organization)
-        elif self.user:
-            queryset = queryset.filter(user=self.user)
-        else:
-            queryset = queryset.filter(organization__isnull=True, user__isnull=True)
+    @classmethod
+    def lookup(cls, url, owner=None):
+        owner_entries = cls.get_active_entries(owner) if owner else cls.objects.none()
+        entry = owner_entries.filter(url=url).first()
+        if not entry:
+            entry = cls.get_active_global_entries().filter(url=url).first()
+        if entry:
+            return ConceptContainerModel.resolve_reference_expression(url=entry.url, namespace=entry.namespace)
 
-        return queryset
+        return None

core/url_registry/tests.py

@@ -1,7 +1,15 @@
-from core.common.tests import OCLTestCase
+from mock.mock import patch
+from rest_framework.exceptions import ErrorDetail
+
+from core.common.tests import OCLTestCase, OCLAPITestCase
 from core.orgs.models import Organization
+from core.orgs.tests.factories import OrganizationFactory
+from core.repos.serializers import RepoListSerializer
+from core.sources.tests.factories import OrganizationSourceFactory
+from core.url_registry.factories import OrganizationURLRegistryFactory, UserURLRegistryFactory, GlobalURLRegistryFactory
 from core.url_registry.models import URLRegistry
 from core.users.models import UserProfile
+from core.users.tests.factories import UserProfileFactory
 
 
 class URLRegistryTest(OCLTestCase):
@@ -25,3 +33,69 @@ def test_owner_url(self):
         self.assertEqual(URLRegistry().owner_url, '/')
         self.assertEqual(URLRegistry(organization=org).owner_url, '/orgs/foo/')
         self.assertEqual(URLRegistry(user=user).owner_url, '/users/foo/')
+
+    @patch('core.common.models.ConceptContainerModel.resolve_reference_expression')
+    def test_lookup(self, resolve_mock):
+        resolve_mock.return_value = 'something'
+
+        org = OrganizationFactory()
+        user = UserProfileFactory()
+        OrganizationURLRegistryFactory(url='https://foo.com', namespace='org1', organization=org)
+        OrganizationURLRegistryFactory(url='https://foo1.com', namespace='org2', organization=org)
+        UserURLRegistryFactory(url='https://foo.com', user=user, namespace='user1')
+        UserURLRegistryFactory(url='https://foo2.com', user=user, namespace='user2')
+        GlobalURLRegistryFactory(url='https://foo.com', namespace='global1')
+        GlobalURLRegistryFactory(url='https://foo1.com', namespace='global2')
+
+        self.assertEqual(URLRegistry.lookup('https://foo.com', org), 'something')
+        resolve_mock.assert_called_with(url='https://foo.com', namespace='org1')
+
+        self.assertEqual(URLRegistry.lookup('https://foo.com'), 'something')
+        resolve_mock.assert_called_with(url='https://foo.com', namespace='global1')
+
+        self.assertEqual(URLRegistry.lookup('https://foo2.com', org), None)
+
+        self.assertEqual(URLRegistry.lookup('https://foo2.com', user), 'something')
+        resolve_mock.assert_called_with(url='https://foo2.com', namespace='user2')
+
+        self.assertEqual(URLRegistry.lookup('https://foo2.com'), None)
+
+        self.assertEqual(URLRegistry.lookup('https://foo1.com'), 'something')
+        resolve_mock.assert_called_with(url='https://foo1.com', namespace='global2')
+
+        self.assertEqual(URLRegistry.lookup('https://foo1.com', org), 'something')
+        resolve_mock.assert_called_with(url='https://foo1.com', namespace='org2')
+
+
+class URLRegistryLookupViewTest(OCLAPITestCase):
+    @patch('core.url_registry.views.URLRegistry.lookup')
+    def test_get(self, lookup_mock):
+        source = OrganizationSourceFactory()
+        lookup_mock.return_value = source
+        response = self.client.get('/url-registry/$lookup/')
+
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(
+            response.data,
+            {'detail': ErrorDetail(string='url is required in query params', code='bad_request')}
+        )
+
+        response = self.client.get('/url-registry/$lookup/?url=https://foo.com')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data, RepoListSerializer(source).data)
+        lookup_mock.assert_called_with('https://foo.com', None)
+
+        lookup_mock.return_value = None
+        response = self.client.get('/url-registry/$lookup/?url=https://foo.com')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data, None)
+        lookup_mock.assert_called_with('https://foo.com', None)
+
+        lookup_mock.return_value = source
+        response = self.client.get(source.organization.uri + 'url-registry/$lookup/?url=https://foo.com')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data, RepoListSerializer(source).data)
+        lookup_mock.assert_called_with('https://foo.com', source.organization)

core/url_registry/urls.py

@@ -1,8 +1,9 @@
-from django.urls import path
+from django.urls import path, re_path
 
 from core.url_registry import views
 
 urlpatterns = [
     path('', views.URLRegistriesView.as_view(), name='url-registries'),
     path('<int:id>/', views.URLRegistryView.as_view(), name='url-registry'),
+    re_path(r'^\$lookup/$', views.URLRegistryLookupView.as_view(), name='url-registry-lookup'),
 ]

core/url_registry/views.py

@@ -2,10 +2,14 @@
 from drf_yasg.utils import swagger_auto_schema
 from rest_framework.generics import CreateAPIView, RetrieveUpdateDestroyAPIView, get_object_or_404
 from rest_framework.permissions import IsAuthenticatedOrReadOnly
+from rest_framework.response import Response
 
+from core.common.exceptions import Http400
 from core.common.mixins import ListWithHeadersMixin
-from core.common.swagger_parameters import q_param, limit_param, sort_desc_param, sort_asc_param, page_param
+from core.common.swagger_parameters import q_param, limit_param, sort_desc_param, sort_asc_param, page_param, \
+    url_registry_url_lookup_param
 from core.common.views import BaseAPIView
+from core.repos.serializers import RepoListSerializer
 from core.url_registry.documents import URLRegistryDocument
 from core.url_registry.models import URLRegistry
 from core.url_registry.serializers import URLRegistryDetailSerializer
@@ -88,3 +92,22 @@ def get_object(self, queryset=None):
         instance = get_object_or_404(queryset.filter(id=self.kwargs['id']))
         self.check_object_permissions(self.request, instance)
         return instance
+
+
+class URLRegistryLookupView(URLRegistryBaseView):
+    serializer_class = RepoListSerializer
+
+    @swagger_auto_schema(manual_parameters=[url_registry_url_lookup_param])
+    def get(self, request, *args, **kwargs):  # pylint: disable=unused-argument
+        self.set_parent_resource()
+        url = request.query_params.get('url')
+
+        if not url:
+            raise Http400('url is required in query params')
+
+        repo = URLRegistry.lookup(url, self.parent_resource)
+        data = None
+        if repo:
+            data = RepoListSerializer(repo).data
+
+        return Response(data)