Merge pull request #57 from GSA/get-source-by-jobid

Get source by jobid

app/interface.py

@@ -287,6 +287,13 @@ def get_harvest_record_by_source(self, source_id):
                 HarvesterDBInterface._to_dict(rcd) for rcd in harvest_records]
             return harvest_records_data
 
+    def get_source_by_jobid(self, jobid):
+        harvest_job = self.db.query(HarvestJob).filter_by(id=jobid).first()
+        if harvest_job is None:
+            return None
+        else:
+            return HarvesterDBInterface._to_dict(harvest_job.source)
+
     def close(self):
         if hasattr(self.db, "remove"):
             self.db.remove()

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "datagov-harvesting-logic"
-version = "0.3.8"
+version = "0.3.9"
 description = ""
 # authors = [
 #     {name = "Jin Sun", email = "[email protected]"},

tests/database/test_db.py

@@ -110,3 +110,18 @@ def test_delete_harvest_source(interface, source_data):
 
     deleted_source = interface.get_harvest_source(source.id)
     assert deleted_source is None
+
+@pytest.fixture
+def job_data(source_data):
+    return {"status": "new"}
+
+
+def test_harvest_source_by_jobid(interface, source_data, job_data):
+
+    source = interface.add_harvest_source(source_data)
+    job_data["harvest_source_id"] = source.id
+
+    harvest_job = interface.add_harvest_job(job_data)
+    harvest_source = interface.get_source_by_jobid(harvest_job.id)
+
+    assert source.id == harvest_source["id"]