Feat (Att): Support external files/links as attachments

.github/workflows/test.yml

@@ -224,13 +224,6 @@ jobs:
             git push
           fi
 
-      - name: Test Weblate API addon access
-        env:
-          WEBLATE_API_TOKEN: ${{ secrets.WEBLATE_API_TOKEN }}
-        run: |
-          curl --fail-with-body --show-error \
-            -H "Authorization: Token $WEBLATE_API_TOKEN" \
-            https://hosted.weblate.org/api/addons/34603/
       # This step must be run after pushing to github or else it will fail in
       # cases when new components are to be created (Weblate requires files to
       # be pushed before component can be created)

specifyweb/backend/stored_queries/queryfieldspec.py

@@ -3,7 +3,9 @@
 from datetime import datetime, timedelta
 import re
 from collections import namedtuple, deque
+from types import SimpleNamespace
 from typing import TypedDict, Iterable
+from urllib.parse import urlencode
 
 from specifyweb.backend.inheritance.utils import get_cat_num_inheritance_setting, get_parent_cat_num_inheritance_setting
 from specifyweb.specify.utils.uiformatters import get_uiformatter
@@ -553,6 +555,7 @@ def add_to_query(
     def add_spec_to_query(
         self, query, formatter=None, aggregator=None, cycle_detector=[]
     ):
+        """Build the shared field projection used by query results and exports."""
 
         if self.get_field() is None:
             return (
@@ -602,6 +605,21 @@ def add_spec_to_query(
                     else:
                         raise
 
+                if table.name == "Attachment" and field_name == "resourceURL":
+                    from specifyweb.backend.attachment_gw import views as attachment_gw
+
+                    if attachment_gw.server_urls is not None:
+                        read_url = attachment_gw.server_urls["read"]
+                        collection_name = attachment_gw.get_collection(
+                            SimpleNamespace(specify_collection=query.collection)
+                        )
+                        params = urlencode({"coll": collection_name, "type": "O"})
+                        separator = "&" if "?" in read_url else "?"
+                        prefix = f"{read_url}{separator}{params}&filename="
+                        orm_field = sql.func.coalesce(
+                            orm_field, prefix + orm_model.attachmentLocation
+                        )
+
                 if field.is_temporal() and self.date_part != "Full Date":
                     precision_field_name = field.name + "Precision"
                     precision_field = getattr(orm_model, precision_field_name, None)

specifyweb/backend/stored_queries/tests/test_attachment_resource_url.py

@@ -0,0 +1,169 @@
+import csv
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from django.test import SimpleTestCase, override_settings
+from sqlalchemy import String, create_engine, sql
+
+from specifyweb.backend.attachment_gw import views as attachment_gw
+from specifyweb.backend.stored_queries.execution import execute, query_to_csv
+from specifyweb.backend.stored_queries.queryfieldspec import QueryFieldSpec
+from specifyweb.backend.stored_queries.tests.tests import SQLAlchemySetup
+from specifyweb.backend.stored_queries.tests.utils import make_query_fields_test
+from specifyweb.specify.models import Attachment
+
+
+@override_settings(WEB_ATTACHMENT_COLLECTION="Test Collection & Images")
+class TestAttachmentResourceURLProjection(SimpleTestCase):
+    def project(self, resource_url, location, path=None):
+        spec = QueryFieldSpec.from_path(path or ["Attachment", "resourceURL"])
+        attachment = sql.select([
+            sql.literal(resource_url, type_=String).label("resourceURL"),
+            sql.literal(location, type_=String).label("attachmentLocation"),
+        ]).alias("attachment")
+        orm_model = SimpleNamespace(**attachment.c)
+        query = Mock()
+        query.collection = SimpleNamespace(collectionname="Separate Collection")
+        query.build_join.return_value = (
+            query, orm_model, spec.table, spec.get_field()
+        )
+        returned_query, projection, field, table = spec.add_spec_to_query(query)
+        self.assertIs(returned_query, query)
+        self.assertIs(field, spec.get_field())
+        self.assertIs(table, spec.table)
+
+        engine = create_engine("sqlite://")
+        try:
+            with engine.connect() as connection:
+                return connection.execute(sql.select([projection])).scalar()
+        finally:
+            engine.dispose()
+
+    def test_projection(self):
+        with patch.object(
+            attachment_gw, "server_urls", {"read": "https://assets.example/fileget"}
+        ):
+            for resource_url, location, expected in [
+                ("https://external.example/image.jpg", "image.jpg",
+                 "https://external.example/image.jpg"),
+                ("https://external.example/image.jpg", None,
+                 "https://external.example/image.jpg"),
+                ("", "image.jpg", ""),
+                (None, "image.jpg",
+                 "https://assets.example/fileget"
+                 "?coll=Test+Collection+%26+Images&type=O&filename=image.jpg"),
+                (None, None, None),
+            ]:
+                with self.subTest(resource_url=resource_url, location=location):
+                    self.assertEqual(self.project(resource_url, location), expected)
+
+    def test_unconfigured_asset_server(self):
+        with patch.object(attachment_gw, "server_urls", None):
+            self.assertEqual(
+                self.project("https://external.example/image.jpg", None),
+                "https://external.example/image.jpg",
+            )
+            self.assertIsNone(self.project(None, "image.jpg"))
+
+    def test_joined_attachment(self):
+        with patch.object(
+            attachment_gw, "server_urls", {"read": "https://assets.example/fileget"}
+        ):
+            self.assertEqual(
+                self.project(None, "joined.jpg", [
+                    "CollectionObject", "collectionObjectAttachments",
+                    "attachment", "resourceURL"
+                ]),
+                "https://assets.example/fileget"
+                "?coll=Test+Collection+%26+Images&type=O&filename=joined.jpg",
+            )
+
+    def test_read_endpoint_with_query_parameters(self):
+        with patch.object(attachment_gw, "server_urls", {
+            "read": "https://assets.example/fileget?existing=value"
+        }):
+            self.assertEqual(
+                self.project(None, "image.jpg"),
+                "https://assets.example/fileget?existing=value"
+                "&coll=Test+Collection+%26+Images&type=O&filename=image.jpg",
+            )
+
+    def test_other_attachment_fields_are_unchanged(self):
+        with patch.object(
+            attachment_gw, "server_urls", {"read": "https://assets.example/fileget"}
+        ):
+            self.assertEqual(
+                self.project(None, "image.jpg", ["Attachment", "attachmentLocation"]),
+                "image.jpg",
+            )
+
+    @override_settings(
+        WEB_ATTACHMENT_COLLECTION=None, SEPARATE_WEB_ATTACHMENT_FOLDERS=True
+    )
+    def test_separate_collection_folder(self):
+        with patch.object(
+            attachment_gw, "server_urls", {"read": "https://assets.example/fileget"}
+        ):
+            self.assertEqual(
+                self.project(None, "image.jpg"),
+                "https://assets.example/fileget"
+                "?coll=Separate+Collection&type=O&filename=image.jpg",
+            )
+
+
+@override_settings(WEB_ATTACHMENT_COLLECTION="Test Collection")
+class TestAttachmentResourceURLResults(SQLAlchemySetup):
+    def setUp(self):
+        super().setUp()
+        server_urls = patch.object(
+            attachment_gw, "server_urls", {"read": "https://assets.example/fileget"}
+        )
+        server_urls.start()
+        self.addCleanup(server_urls.stop)
+        self.attachments = [
+            Attachment.objects.create(resourceurl=url, attachmentlocation=location)
+            for url, location in [
+                ("https://external.example/image.jpg", "external.jpg"),
+                (None, "image.jpg"),
+                ("", "blank.jpg"),
+                (None, None),
+            ]
+        ]
+        self.expected = [
+            (self.attachments[0].id, "https://external.example/image.jpg"),
+            (self.attachments[1].id,
+             "https://assets.example/fileget"
+             "?coll=Test+Collection&type=O&filename=image.jpg"),
+            (self.attachments[2].id, ""),
+            (self.attachments[3].id, None),
+        ]
+        self.table, self.fields = make_query_fields_test(
+            "Attachment", [["resourceURL"]]
+        )
+
+    def test_query_results(self):
+        with TestAttachmentResourceURLResults.test_session_context() as session:
+            result = execute(
+                session, self.collection, self.specifyuser, self.table.tableId,
+                search_synonymy=False, distinct=False, series=False,
+                count_only=False, query_fields=self.fields, limit=0, offset=0,
+            )
+        self.assertCountEqual(result["results"], self.expected)
+
+    def test_csv_export(self):
+        with TemporaryDirectory() as directory:
+            path = Path(directory) / "attachments.csv"
+            with TestAttachmentResourceURLResults.test_session_context() as session:
+                query_to_csv(
+                    session, self.collection, self.specifyuser,
+                    self.table.tableId, self.fields, path,
+                )
+            with path.open(newline="") as exported:
+                rows = list(csv.reader(exported))
+        self.assertCountEqual(
+            rows,
+            [[str(record_id), url if url is not None else ""]
+             for record_id, url in self.expected],
+        )

specifyweb/backend/workbench/upload/parse.py

@@ -6,6 +6,9 @@
 from datetime import datetime
 from decimal import Decimal
 
+from django.core.exceptions import ValidationError
+from django.core.validators import URLValidator
+
 from specifyweb.specify.utils.agent_types import agent_types
 from specifyweb.backend.stored_queries.format import get_date_format, MYSQL_TO_YEAR, MYSQL_TO_MONTH
 from specifyweb.specify.datamodel import datamodel, Table, Field, Relationship
@@ -19,6 +22,7 @@
     'failedParsingFloat',
     'failedParsingBoolean',
     'failedParsingAgentType',
+    'failedParsingURL',
 
     'invalidYear',
     'badDateFormat',
@@ -43,6 +47,8 @@ class ParseSucess(NamedTuple):
 
 ParseResult = ParseSucess | ParseFailure
 
+validate_resource_url = URLValidator(schemes=['http', 'https'])
+
 
 def parse_field(
         table_name: str,
@@ -57,7 +63,14 @@ def parse_field(
         return parse_integer(field.name, raw_value)
 
     if formatter is not None:
-        return parse_formatted(formatter, table, field, raw_value)
+        result = parse_formatted(formatter, table, field, raw_value)
+        if (
+            table.name == 'Attachment'
+            and field.name.lower() == 'resourceurl'
+            and isinstance(result, ParseSucess)
+        ):
+            return parse_field(table_name, field_name, result.payload[field.name.lower()])
+        return result
 
     if is_latlong(table, field):
         return parse_latlong(field, raw_value)
@@ -84,6 +97,11 @@ def parse_field(
     if hasattr(field, 'length') and field.length is not None and len(raw_value) > field.length:
         return ParseFailure('valueTooLong', {'field': field_name, 'maxLength': field.length})
 
+    if table.name == 'Attachment' and field.name.lower() == 'resourceurl':
+        try:
+            validate_resource_url(raw_value)
+        except ValidationError:
+            return ParseFailure('failedParsingURL', {'value': raw_value})
 
     return ParseSucess({field_name.lower(): raw_value})
 

specifyweb/backend/workbench/upload/parsing.py

@@ -117,6 +117,14 @@ def _parse(tablename: str, fieldname: str, colopts: ExtendedColumnOptions, value
                     },
                     colopts.column
                 )
+            if (
+                isinstance(result, ParseResult)
+                and table.name == 'Attachment'
+                and field.name.lower() == 'resourceurl'
+            ):
+                parsed_url = parse_field(tablename, fieldname, result.upload[fieldname])
+                if isinstance(parsed_url, ParseFailure):
+                    return WorkBenchParseFailure.from_parse_failure(parsed_url, colopts.column)
             return result
 
     parsed = parse_field(tablename, fieldname, value, colopts.uiformatter)