fixes count decorator to work on queries where from is not supplied b…

…y ORM;

fixes job errors subquery to filter by job

database/interface.py

@@ -44,12 +44,18 @@ def _impl(self, *args, **kwargs):
     return _impl
 
 
+# notes on the flag `maintain_column_froms`:
+# https://github.com/sqlalchemy/sqlalchemy/discussions/6807#discussioncomment-1043732
+# docs: https://docs.sqlalchemy.org/en/14/core/selectable.html#sqlalchemy.sql.expression.Select.with_only_columns.params.maintain_column_froms
+#
 def count(fn):
     @wraps(fn)
     def _impl(self, *args, **kwargs):
         query = fn(self, *args, **kwargs)
         if kwargs.get("count") is True:
-            count_q = query.statement.with_only_columns(*[func.count()]).order_by(None)
+            count_q = query.statement.with_only_columns(
+                func.count(), maintain_column_froms=True
+            ).order_by(None)
             count = query.session.execute(count_q).scalar()
             return count
         else:
@@ -421,6 +427,7 @@ def get_harvest_record_errors_by_job(self, job_id: str, **kwargs):
         subquery = (
             self.db.query(HarvestRecord.id)
             .filter(HarvestRecord.status == "error")
+            .filter(HarvestRecord.harvest_job_id == job_id)
             .subquery()
         )
         query = self.db.query(HarvestRecordError).filter(
@@ -604,19 +611,22 @@ def verify_user(self, usr_data):
             return False
 
     #### PAGINATED QUERIES
-
+    @count
     @paginate
     def pget_harvest_jobs(self, filter=text(""), **kwargs):
         return self.db.query(HarvestJob).filter(filter)
 
+    @count
     @paginate
     def pget_harvest_records(self, filter=text(""), **kwargs):
         return self.db.query(HarvestRecord).filter(filter)
 
+    @count
     @paginate
     def pget_harvest_job_errors(self, filter=text(""), **kwargs):
         return self.db.query(HarvestJobError).filter(filter)
 
+    @count
     @paginate
     def pget_harvest_record_errors(self, filter=text(""), **kwargs):
         return self.db.query(HarvestRecordError).filter(filter)

tests/conftest.py

@@ -220,11 +220,11 @@ def job_data_dcatus_orm(job_data_dcatus: dict) -> HarvestJob:
 
 
 @pytest.fixture
-def job_data_dcatus_2(source_data_dcatus: dict) -> dict:
+def job_data_dcatus_2(source_data_dcatus_2: dict) -> dict:
     return {
         "id": "392ac4b3-79a6-414b-a2b3-d6c607d3b8d4",
         "status": "new",
-        "harvest_source_id": source_data_dcatus["id"],
+        "harvest_source_id": source_data_dcatus_2["id"],
     }
 
 
@@ -278,12 +278,43 @@ def record_data_dcatus(fixtures_json) -> List[dict]:
     return fixtures_json["record"]
 
 
+@pytest.fixture
+def record_data_dcatus_2(job_data_dcatus_2):
+    return [
+        {
+            "id": "72bae4b2-336e-49df-bc4c-410dc73dc316",
+            "identifier": "test_identifier-2",
+            "harvest_job_id": job_data_dcatus_2["id"],
+            "harvest_source_id": job_data_dcatus_2["harvest_source_id"],
+            "action": "create",
+            "status": "error",
+            "source_raw": "example data 2",
+        }
+    ]
+
+
 ## HARVEST RECORD ERRORS
 @pytest.fixture
 def record_error_data(fixtures_json) -> List[dict]:
     return fixtures_json["record_error"]
 
 
+@pytest.fixture
+def record_error_data_2(record_data_dcatus_2) -> dict:
+    return [
+        {
+            "harvest_record_id": record_data_dcatus_2[0]["id"],
+            "message": "record is invalid",
+            "type": "ValidationException",
+        },
+        {
+            "harvest_record_id": record_data_dcatus_2[0]["id"],
+            "message": "record is invalid_2",
+            "type": "ValidationException",
+        },
+    ]
+
+
 @pytest.fixture
 def interface_no_jobs(interface, organization_data, source_data_dcatus):
     interface.add_organization(organization_data)
@@ -310,6 +341,24 @@ def interface_with_fixture_json(
     return interface_no_jobs
 
 
+@pytest.fixture
+def interface_with_multiple_sources(
+    interface_with_fixture_json,
+    source_data_dcatus_2,
+    job_data_dcatus_2,
+    record_data_dcatus_2,
+    record_error_data_2,
+):
+    interface_with_fixture_json.add_harvest_source(source_data_dcatus_2)
+    interface_with_fixture_json.add_harvest_job(job_data_dcatus_2)
+    for record in record_data_dcatus_2:
+        interface_with_fixture_json.add_harvest_record(record)
+    for error in record_error_data_2:
+        interface_with_fixture_json.add_harvest_record_error(error)
+
+    return interface_with_fixture_json
+
+
 ## MISC
 @pytest.fixture
 def interface_with_multiple_jobs(interface_no_jobs, source_data_dcatus):

tests/integration/database/test_db.py

@@ -276,12 +276,32 @@ def test_endpoint_count(
         self, interface_with_fixture_json, job_data_dcatus, record_data_dcatus
     ):
         interface = interface_with_fixture_json
-        job_id = job_data_dcatus
+        job_id = job_data_dcatus["id"]
         count = interface.get_harvest_record_errors_by_job(
             job_id, count=True, skip_pagination=True
         )
         assert count == len(record_data_dcatus)
 
+    def test_errors_by_job(
+        self,
+        interface_with_multiple_sources,
+        job_data_dcatus,
+        job_data_dcatus_2,
+        record_error_data,
+        record_error_data_2,
+    ):
+        interface = interface_with_multiple_sources
+        job_id = job_data_dcatus["id"]
+        count = interface.get_harvest_record_errors_by_job(
+            job_id, count=True, skip_pagination=True
+        )
+        all_errors_count = interface.pget_harvest_record_errors(
+            count=True,
+            skip_pagination=True,
+        )
+        assert count == len(record_error_data)
+        assert all_errors_count == len(record_error_data) + len(record_error_data_2)
+
     def test_add_harvest_job_with_id(
         self, interface, organization_data, source_data_dcatus, job_data_dcatus
     ):