Remove pandas from gsaid api

cg/cli/upload/fohm.py

@@ -79,7 +79,7 @@ def preprocess_all(
     upload_cases = []
     for case_id in cases:
         try:
-            gisaid_api.upload(case_id=case_id)
+            gisaid_api.upload_to_gisaid(case_id)
             fohm_api.update_upload_started_at(case_id=case_id)
             LOG.info(f"Upload of case {case_id} to GISAID was successful")
             upload_cases.append(case_id)

cg/cli/upload/gisaid.py

@@ -1,4 +1,4 @@
-"""Code for uploading genotype data via CLI"""
+"""Code for uploading genotype data via CLI."""
 
 import logging
 
@@ -18,7 +18,7 @@ def upload_to_gisaid(context: CGConfig, case_id: str):
 
     LOG.info("----------------- GISAID UPLOAD -------------------")
 
-    gisaid_api = GisaidAPI(config=context)
+    gisaid_api = GisaidAPI(context)
 
-    gisaid_api.upload(case_id=case_id)
+    gisaid_api.upload_to_gisaid(case_id)
     LOG.info("Upload to GISAID successful")

cg/constants/constants.py

@@ -200,6 +200,7 @@ class FileExtensions(StrEnum):
     CONFIG: str = ".config"
     CRAM: str = ".cram"
     CSV: str = ".csv"
+    FASTA: str = ".fasta"
     FASTQ: str = ".fastq"
     FASTQ_GZ: str = ".fastq.gz"
     GPG: str = ".gpg"

cg/constants/housekeeper_tags.py

@@ -250,3 +250,10 @@ class JanusTags:
 class FohmTag(StrEnum):
     COMPLEMENTARY = "komplettering"
     PANGOLIN_TYPING = "pangolin-typing-fohm"
+
+
+class GisaidTag(StrEnum):
+    FASTA = "gisaid-fasta"
+    CONSENSUS_SAMPLE = "consensus-sample"
+    CSV = "gisaid-csv"
+    LOG = "gisaid-log"

cg/models/gisaid/reports.py

@@ -0,0 +1,11 @@
+"""GISAID report models."""
+
+from pydantic import BaseModel, Field
+
+
+class GisaidComplementaryReport(BaseModel):
+    """Model for validating a GSAID complementary reports."""
+
+    gisaid_accession: str = Field(None, alias="GISAID_accession")
+    sample_number: str = Field(str, alias="provnummer")
+    selection_criteria: str = Field(str, alias="urvalskriterium")

pyproject.toml

@@ -57,7 +57,6 @@ marshmallow = "*"
 MarkupSafe = "*"
 openpyxl = "*"
 packaging = "*"
-pandas = "*"
 paramiko = "*"
 petname = "*"
 psutil = "*"

tests/conftest.py

@@ -118,6 +118,8 @@
     "tests.fixture_plugins.encryption_fixtures.encryption_fixtures",
     "tests.fixture_plugins.fohm.fohm_fixtures",
     "tests.fixture_plugins.io.csv_fixtures",
+    "tests.fixture_plugins.io.csv_fixtures",
+    "tests.fixture_plugins.gisaid_fixtures.gisaid_fixtures",
     "tests.fixture_plugins.illumina_clean_fixtures.clean_fixtures",
     "tests.fixture_plugins.loqusdb_fixtures.loqusdb_api_fixtures",
     "tests.fixture_plugins.loqusdb_fixtures.loqusdb_output_fixtures",

tests/fixture_plugins/gisaid_fixtures/gisaid_fixtures.py

@@ -0,0 +1,47 @@
+import pytest
+
+from cg.meta.upload.gisaid import GisaidAPI
+from cg.models.cg_config import CGConfig
+from cg.models.gisaid.reports import GisaidComplementaryReport
+
+
+@pytest.fixture
+def gisaid_complementary_report_raw() -> dict[str, str]:
+    """Return a raw GISAID complementary report."""
+    return {
+        "provnummer": "a_sample_number",
+        "urvalskriterium": "a_selection_criteria",
+    }
+
+
+@pytest.fixture
+def gisaid_complementary_report(
+    gisaid_complementary_report_raw: list[dict],
+) -> GisaidComplementaryReport:
+    """Return GisaidComplementaryReport."""
+    return GisaidComplementaryReport.model_validate(gisaid_complementary_report_raw)
+
+
+@pytest.fixture
+def gisaid_complementary_reports(
+    gisaid_complementary_report: GisaidComplementaryReport,
+) -> list[GisaidComplementaryReport]:
+    """Return GISAID complementary reports."""
+    report_1 = GisaidComplementaryReport.model_validate(
+        {"provnummer": "1CS", "urvalskriterium": "criteria", "GISAID_accession": "an_accession"}
+    )
+    complementary_report = GisaidComplementaryReport.model_validate(
+        {
+            "urvalskriterium": "criteria",
+            "provnummer": "44CS000000",
+        }
+    )
+    return [gisaid_complementary_report, complementary_report, report_1]
+
+
+@pytest.fixture
+def gisaid_api(
+    cg_context: CGConfig,
+) -> GisaidAPI:
+    """GISAID API fixture."""
+    return GisaidAPI(cg_context)

tests/meta/upload/gisaid/test_gsaid.py

@@ -0,0 +1,82 @@
+from pathlib import Path
+
+from cg.meta.upload.gisaid import GisaidAPI
+from cg.models.gisaid.reports import GisaidComplementaryReport
+
+
+def test_get_complementary_report_content(gisaid_api: GisaidAPI, csv_file_path: Path):
+    # GIVEN a list of CSV files
+
+    # WHEN creating the report content
+    content: list[dict] = gisaid_api.get_complementary_report_content(csv_file_path)
+
+    # THEN each file is a list of dicts where each dict represents a row in a CSV file
+    assert isinstance(content[0], dict)
+
+    # THEN the file is added as a list of dicts
+    assert len(content) == 3
+
+
+def test_validate_gisaid_complementary_reports(
+    gisaid_api: GisaidAPI, gisaid_complementary_report_raw: dict[str, str]
+):
+    # GIVEN a dict
+
+    # WHEN validating the dict
+    content: list[GisaidComplementaryReport] = gisaid_api.validate_gisaid_complementary_reports(
+        [gisaid_complementary_report_raw]
+    )
+
+    # THEN a list of reports is returned
+    assert isinstance(content[0], GisaidComplementaryReport)
+
+
+def test_get_sars_cov_complementary_reports(
+    gisaid_complementary_reports: list[GisaidComplementaryReport], gisaid_api: GisaidAPI
+):
+    # GIVEN a list of reports
+
+    # WHEN getting Sars-cov reports from reports
+    content: list[GisaidComplementaryReport] = gisaid_api.get_sars_cov_complementary_reports(
+        gisaid_complementary_reports
+    )
+
+    # THEN a list of reports is returned
+    assert isinstance(content[0], GisaidComplementaryReport)
+
+    # THEN only the report for Sars-cov2 reports remains
+    assert len(content) == 1
+    assert content[0].sample_number == "44CS000000"
+
+
+def test_get_complementary_report_sample_number(
+    gisaid_complementary_reports: list[GisaidComplementaryReport], gisaid_api: GisaidAPI
+):
+    # GIVEN a list of reports
+
+    # WHEN getting the sample numbers in the reports
+    sample_numbers: set[str] = gisaid_api.get_complementary_report_sample_number(
+        gisaid_complementary_reports
+    )
+
+    # THEN return sample numbers from reports
+    for report in gisaid_complementary_reports:
+        assert report.sample_number in sample_numbers
+
+
+def test_add_gisaid_accession_to_reports(
+    gisaid_complementary_reports: list[GisaidComplementaryReport], gisaid_api: GisaidAPI
+):
+    """Test adding gisaid accession to the reports."""
+    # GIVEN a GISAID API
+
+    # GIVEN a list of reports
+
+    # WHEN adding GISAID accession to reports
+    gisaid_api.add_gisaid_accession_to_complementary_reports(
+        gisaid_accession={gisaid_complementary_reports[0].sample_number: "a_gisaid_accession"},
+        reports=[gisaid_complementary_reports[0]],
+    )
+
+    # THEN a GISAID accession has been added
+    assert isinstance(gisaid_complementary_reports[0].gisaid_accession, str)