✨ Add instrument methods (#282)

.github/workflows/ci.yml

@@ -37,7 +37,7 @@ jobs:
         poetry install -E data_science
     - name: Run doctests
       # Forks can't run doctests, requires super user token
-      if: github.actor == 'pwildenhain'
+      if: github.triggering_actor == 'pwildenhain'
       run: |
         poetry run pytest --doctest-only --doctest-plus
     - name: Run tests

docs/quickstart.md

@@ -35,7 +35,7 @@ Export a file
 
 ```python
 content, headers = project.export_file('1', 'file')
-with open(headers['name'], 'w') as fobj:
+with open(headers['name'], 'wb') as fobj:
     fobj.write(content)
 ```
 
@@ -50,3 +50,11 @@ except ValueError:
     # You screwed up and gave it a bad field name, etc
     pass
 ```
+
+Export a PDF file of all instruments (blank)
+
+```python
+content, _headers = project.export_pdf()
+with open('all_instruments_blank.pdf', 'wb') as fobj:
+    fobj.write(content)
+```

redcap/methods/base.py

@@ -341,6 +341,7 @@ def _return_data(
             "event",
             "exportFieldNames",
             "formEventMapping",
+            "instrument",
             "log",
             "metadata",
             "participantList",

redcap/methods/instruments.py

@@ -1,16 +1,7 @@
 """REDCap API methods for Project instruments"""
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Dict,
-    List,
-    Literal,
-    Optional,
-    Union,
-    cast,
-)
-
-from redcap.methods.base import Base
+from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Union, cast
+
+from redcap.methods.base import Base, FileMap
 from redcap.request import Json
 
 if TYPE_CHECKING:
@@ -20,6 +11,116 @@
 class Instruments(Base):
     """Responsible for all API methods under 'Instruments' in the API Playground"""
 
+    def export_instruments(
+        self,
+        format_type: Literal["json", "csv", "xml", "df"] = "json",
+    ):
+        """
+        Export the Instruments of the Project
+
+        Args:
+            format_type:
+                Response return format
+
+        Returns:
+            Union[List[Dict[str, Any]], str, pandas.DataFrame]: List of Instruments
+
+        Examples:
+            >>> proj.export_instruments()
+            [{'instrument_name': 'form_1', 'instrument_label': 'Form 1'}]
+        """
+        payload = self._initialize_payload(
+            content="instrument", format_type=format_type
+        )
+        return_type = self._lookup_return_type(format_type, request_type="export")
+        response = cast(Union[Json, str], self._call_api(payload, return_type))
+
+        return self._return_data(
+            response=response,
+            content="instrument",
+            format_type=format_type,
+        )
+
+    #### pylint: disable=too-many-locals
+
+    def export_pdf(
+        self,
+        record: Optional[str] = None,
+        event: Optional[str] = None,
+        instrument: Optional[str] = None,
+        repeat_instance: Optional[int] = None,
+        all_records: Optional[bool] = None,
+        compact_display: Optional[bool] = None,
+    ) -> FileMap:
+        """
+        Export PDF file of instruments, either as blank or with data
+
+        Args:
+            record: Record ID
+            event: For longitudinal projects, the unique event name
+            instrument: Unique instrument name
+            repeat_instance:
+                (Only for projects with repeating instruments/events)
+                The repeat instance number of the repeating event (if longitudinal)
+                or the repeating instrument (if classic or longitudinal).
+            all_records:
+                If True, then all records will be exported as a single PDF file.
+                Note: If this is True, then record, event, and instrument parameters
+                      are all ignored.
+            compact_display:
+                If True, then the PDF will be exported in compact display mode.
+
+        Returns:
+            Content of the file and dictionary of useful metadata
+
+        Examples:
+            >>> proj.export_pdf()
+            (b'%PDF-1.3\\n3 0 obj\\n..., {...})
+        """
+        # load up payload
+        payload = self._initialize_payload(content="pdf", return_format_type="json")
+        keys_to_add = (
+            record,
+            event,
+            instrument,
+            repeat_instance,
+            all_records,
+            compact_display,
+        )
+        str_keys = (
+            "record",
+            "event",
+            "instrument",
+            "repeat_instance",
+            "allRecords",
+            "compactDisplay",
+        )
+        for key, data in zip(str_keys, keys_to_add):
+            data = cast(str, data)
+            if data:
+                payload[key] = data
+        payload["action"] = "export"
+
+        content, headers = cast(
+            FileMap, self._call_api(payload=payload, return_type="file_map")
+        )
+        # REDCap adds some useful things in content-type
+        content_map = {}
+        if "content-type" in headers:
+            splat = [
+                key_values.strip() for key_values in headers["content-type"].split(";")
+            ]
+            key_values = [
+                (key_values.split("=")[0], key_values.split("=")[1].replace('"', ""))
+                for key_values in splat
+                if "=" in key_values
+            ]
+            content_map = dict(key_values)
+
+        return content, content_map
+
+    #### pylint: enable=too-many-locals
+
     def export_instrument_event_mappings(
         self,
         format_type: Literal["json", "csv", "xml", "df"] = "json",
@@ -62,3 +163,51 @@ def export_instrument_event_mappings(
             format_type=format_type,
             df_kwargs=df_kwargs,
         )
+
+    def import_instrument_event_mappings(
+        self,
+        to_import: Union[str, List[Dict[str, Any]], "pd.DataFrame"],
+        return_format_type: Literal["json", "csv", "xml"] = "json",
+        import_format: Literal["json", "csv", "xml", "df"] = "json",
+    ):
+        # pylint: disable=line-too-long
+        """
+        Import the project's instrument to event mapping
+
+        Note:
+            This only works for longitudinal projects.
+
+        Args:
+            to_import: array of dicts, csv/xml string, `pandas.DataFrame`
+                Note:
+                    If you pass a csv or xml string, you should use the
+                    `import format` parameter appropriately.
+            return_format_type:
+                Response format. By default, response will be json-decoded.
+            import_format:
+                Format of incoming data. By default, import_format
+                will be json-encoded
+
+        Returns:
+            Union[int, str]: Number of instrument-event mappings imported
+
+        Examples:
+            Import instrument-event mappings
+            >>> instrument_event_mappings = [{"arm_num": "1", "unique_event_name": "event_1_arm_1", "form": "form_1"}]
+            >>> proj.import_instrument_event_mappings(instrument_event_mappings)
+            1
+        """
+        payload = self._initialize_import_payload(
+            to_import=to_import,
+            import_format=import_format,
+            return_format_type=return_format_type,
+            content="formEventMapping",
+        )
+        payload["action"] = "import"
+
+        return_type = self._lookup_return_type(
+            format_type=return_format_type, request_type="import"
+        )
+        response = cast(Union[Json, str], self._call_api(payload, return_type))
+
+        return response

tests/integration/test_long_project.py

@@ -184,8 +184,12 @@ def test_arms_delete(long_project):
 
 @pytest.mark.integration
 def test_arms_import_override(long_project):
-    # Cache current events, so they can be restored for subsequent tests
-    current_events = long_project.export_events()
+    # Cache current events, so they can be restored for subsequent tests, because arms, events,
+    # and mappings are deleted when the 'override' parameter is used.
+    state_dict = {
+        "events": long_project.export_events(),
+        "form_event_map": long_project.export_instrument_event_mappings(),
+    }
 
     new_arms = [{"arm_num": 3, "name": "Drug C"}]
     response = long_project.import_arms(new_arms)
@@ -206,9 +210,14 @@ def test_arms_import_override(long_project):
     with pytest.raises(RedcapError):
         response = long_project.export_arms()
 
-    response = long_project.import_events(current_events)
+    response = long_project.import_events(state_dict["events"])
     assert response == 16
 
+    response = long_project.import_instrument_event_mappings(
+        state_dict["form_event_map"]
+    )
+    assert response == 44
+
     response = long_project.export_arms()
     assert len(response) == 2
 
@@ -248,3 +257,53 @@ def test_events_delete(long_project):
     response = long_project.export_events()
 
     assert len(response) == 16
+
+
+@pytest.mark.integration
+def test_export_instruments(long_project):
+    response = long_project.export_instruments()
+    assert len(response) == 9
+
+
+@pytest.mark.integration
+def test_export_pdf(long_project):
+    content, _ = long_project.export_pdf()
+
+    assert isinstance(content, bytes)
+
+
+@pytest.mark.integration
+def test_fem_export(long_project):
+    response = long_project.export_instrument_event_mappings()
+
+    assert len(response) == 44
+
+
+@pytest.mark.integration
+def test_fem_import(long_project):
+    # Cache current instrument-event mappings, so they can be restored for subsequent tests
+    current_fem = long_project.export_instrument_event_mappings()
+
+    instrument_event_mappings = [
+        {
+            "arm_num": "1",
+            "unique_event_name": "enrollment_arm_1",
+            "form": "demographics",
+        }
+    ]
+    response = long_project.import_instrument_event_mappings(instrument_event_mappings)
+    assert response == 1
+
+    response = long_project.export_instrument_event_mappings()
+    assert len(response) == 1
+
+    fem_arm_nums = [fem["arm_num"] for fem in response]
+    fem_unique_event_names = [fem["unique_event_name"] for fem in response]
+    fem_forms = [fem["form"] for fem in response]
+
+    assert fem_arm_nums == [1]
+    assert fem_unique_event_names == ["enrollment_arm_1"]
+    assert fem_forms == ["demographics"]
+
+    response = long_project.import_instrument_event_mappings(current_fem)
+    assert response == 44

tests/integration/test_simple_project.py

@@ -1,7 +1,6 @@
 """Test suite for simple REDCap Project against real REDCap server"""
 # pylint: disable=missing-function-docstring
 import os
-
 from io import StringIO
 
 import pandas as pd
@@ -216,6 +215,19 @@ def test_export_field_names_df(simple_project):
     assert all(field_names.columns == ["choice_value", "export_field_name"])
 
 
+@pytest.mark.integration
+def test_export_instruments(simple_project):
+    instruments = simple_project.export_instruments()
+    assert len(instruments) == 1
+
+
+@pytest.mark.integration
+def test_export_pdf(simple_project):
+    content, _ = simple_project.export_pdf()
+
+    assert isinstance(content, bytes)
+
+
 @pytest.mark.integration
 def test_export_and_import_metadata(simple_project):
     original_metadata = simple_project.export_metadata()
@@ -279,3 +291,9 @@ def test_export_arms(simple_project):
 def test_export_events(simple_project):
     with pytest.raises(RedcapError):
         simple_project.export_events()
+
+
+@pytest.mark.integration
+def test_export_instrument_event_mapping(simple_project):
+    with pytest.raises(RedcapError):
+        simple_project.export_instrument_event_mappings()