Merge pull request #72 from lindsay-stevens/pyodk-60

60: add form creation

pyodk/_endpoints/entities.py

@@ -1,6 +1,5 @@
 import logging
 from datetime import datetime
-from uuid import uuid4
 
 from pyodk._endpoints import bases
 from pyodk._utils import validators as pv
@@ -121,7 +120,7 @@ def create(
                 entity_list_name, self.default_entity_list_name
             )
             req_data = {
-                "uuid": pv.validate_str(uuid, str(uuid4()), key="uuid"),
+                "uuid": pv.validate_str(uuid, self.session.get_xform_uuid(), key="uuid"),
                 "label": pv.validate_str(label, key="label"),
                 "data": pv.validate_dict(data, key="data"),
             }

pyodk/_endpoints/form_draft_attachments.py

@@ -1,5 +1,5 @@
 import logging
-from pathlib import Path
+from os import PathLike
 
 from pyodk._endpoints import bases
 from pyodk._utils import validators as pv
@@ -34,7 +34,7 @@ def __init__(
 
     def upload(
         self,
-        file_path: str,
+        file_path: PathLike | str,
         file_name: str | None = None,
         form_id: str | None = None,
         project_id: int | None = None,
@@ -50,7 +50,7 @@ def upload(
         try:
             pid = pv.validate_project_id(project_id, self.default_project_id)
             fid = pv.validate_form_id(form_id, self.default_form_id)
-            file_path = Path(pv.validate_file_path(file_path))
+            file_path = pv.validate_file_path(file_path)
             if file_name is None:
                 file_name = pv.validate_str(file_path.name, key="file_name")
         except PyODKError as err:

pyodk/_endpoints/form_drafts.py

@@ -1,13 +1,84 @@
 import logging
-from contextlib import nullcontext
-from pathlib import Path
+from io import BytesIO
+from os import PathLike
+from zipfile import is_zipfile
 
 from pyodk._endpoints import bases
 from pyodk._utils import validators as pv
 from pyodk._utils.session import Session
 from pyodk.errors import PyODKError
 
 log = logging.getLogger(__name__)
+CONTENT_TYPES = {
+    ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+    ".xls": "application/vnd.ms-excel",
+    ".xml": "application/xml",
+}
+
+
+def is_xls_file(buf: bytes) -> bool:
+    """
+    Implements the Microsoft Excel (Office 97-2003) document type matcher.
+
+    From h2non/filetype v1.2.0, MIT License, Copyright (c) 2016 Tomás Aparicio
+
+    :param buf: buffer to match against.
+    """
+    if len(buf) > 520 and buf[0:8] == b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1":
+        if buf[512:516] == b"\xfd\xff\xff\xff" and (buf[518] == 0x00 or buf[518] == 0x02):
+            return True
+        if buf[512:520] == b"\x09\x08\x10\x00\x00\x06\x05\x00":
+            return True
+        if (
+            len(buf) > 2095
+            and b"\xe2\x00\x00\x00\x5c\x00\x70\x00\x04\x00\x00Calc" in buf[1568:2095]
+        ):
+            return True
+
+    return False
+
+
+def get_definition_data(
+    definition: PathLike | str | bytes | None,
+) -> (bytes, str, str | None):
+    """
+    Get the form definition data from a path or bytes.
+
+    :param definition: The path to the file to upload (string or PathLike), or the
+          form definition in memory (string (XML) or bytes (XLS/XLSX)).
+    :return: definition_data, content_type, file_path_stem (if any).
+    """
+    definition_data = None
+    content_type = None
+    file_path_stem = None
+    if (
+        isinstance(definition, str)
+        and """http://www.w3.org/2002/xforms""" in definition[:1000]
+    ):
+        content_type = CONTENT_TYPES[".xml"]
+        definition_data = definition.encode("utf-8")
+    elif isinstance(definition, str | PathLike):
+        file_path = pv.validate_file_path(definition)
+        file_path_stem = file_path.stem
+        definition_data = file_path.read_bytes()
+        if file_path.suffix not in CONTENT_TYPES:
+            raise PyODKError(
+                "Parameter 'definition' file name has an unexpected file extension, "
+                "expected one of '.xlsx', '.xls', '.xml'."
+            )
+        content_type = CONTENT_TYPES[file_path.suffix]
+    elif isinstance(definition, bytes):
+        definition_data = definition
+        if is_zipfile(BytesIO(definition)):
+            content_type = CONTENT_TYPES[".xlsx"]
+        elif is_xls_file(definition):
+            content_type = CONTENT_TYPES[".xls"]
+    if definition_data is None or content_type is None:
+        raise PyODKError(
+            "Parameter 'definition' has an unexpected file type, "
+            "expected one of '.xlsx', '.xls', '.xml'."
+        )
+    return definition_data, content_type, file_path_stem
 
 
 class URLs(bases.Model):
@@ -36,86 +107,79 @@ def __init__(
 
     def _prep_form_post(
         self,
-        file_path: Path | str | None = None,
+        definition: PathLike | str | bytes | None = None,
         ignore_warnings: bool | None = True,
         form_id: str | None = None,
         project_id: int | None = None,
-    ) -> (str, str, dict, dict):
+    ) -> (str, str, dict, dict, bytes | None):
         """
         Prepare / validate input arguments for POSTing a new form definition or version.
 
-        :param file_path: The path to the file to upload.
+        :param definition: The path to the file to upload (string or PathLike), or the
+          form definition in memory (string (XML) or bytes (XLS/XLSX)).
         :param form_id: The xmlFormId of the Form being referenced.
         :param project_id: The id of the project this form belongs to.
         :param ignore_warnings: If True, create the form if there are XLSForm warnings.
         :return: project_id, form_id, headers, params
         """
         try:
             pid = pv.validate_project_id(project_id, self.default_project_id)
-            fid = pv.validate_form_id(form_id, self.default_form_id)
             headers = {}
             params = {}
-            if file_path is not None:
+            definition_data = None
+            file_path_stem = None
+            if definition is not None:
+                definition_data, content_type, file_path_stem = get_definition_data(
+                    definition=definition
+                )
+                headers["Content-Type"] = content_type
+            fid = pv.validate_form_id(
+                form_id,
+                self.default_form_id,
+                file_path_stem,
+                self.session.get_xform_uuid(),
+            )
+            if definition is not None:
                 if ignore_warnings is not None:
                     key = "ignore_warnings"
                     params["ignoreWarnings"] = pv.validate_bool(ignore_warnings, key=key)
-                file_path = Path(pv.validate_file_path(file_path))
-                if file_path.suffix == ".xlsx":
-                    content_type = (
-                        "application/vnd.openxmlformats-"
-                        "officedocument.spreadsheetml.sheet"
-                    )
-                elif file_path.suffix == ".xls":
-                    content_type = "application/vnd.ms-excel"
-                elif file_path.suffix == ".xml":
-                    content_type = "application/xml"
-                else:
-                    raise PyODKError(  # noqa: TRY301
-                        "Parameter 'file_path' file name has an unexpected extension, "
-                        "expected one of '.xlsx', '.xls', '.xml'."
-                    )
-                headers = {
-                    "Content-Type": content_type,
-                    "X-XlsForm-FormId-Fallback": self.session.urlquote(file_path.stem),
-                }
+                headers["X-XlsForm-FormId-Fallback"] = self.session.urlquote(fid)
         except PyODKError as err:
             log.error(err, exc_info=True)
             raise
 
-        return pid, fid, headers, params
+        return pid, fid, headers, params, definition_data
 
     def create(
         self,
-        file_path: Path | str | None = None,
+        definition: PathLike | str | bytes | None = None,
         ignore_warnings: bool | None = True,
         form_id: str | None = None,
         project_id: int | None = None,
     ) -> bool:
         """
         Create a Form Draft.
 
-        :param file_path: The path to the file to upload.
+        :param definition: The path to the file to upload (string or PathLike), or the
+          form definition in memory (string (XML) or bytes (XLS/XLSX)).
         :param form_id: The xmlFormId of the Form being referenced.
         :param project_id: The id of the project this form belongs to.
         :param ignore_warnings: If True, create the form if there are XLSForm warnings.
         """
-        pid, fid, headers, params = self._prep_form_post(
-            file_path=file_path,
+        pid, fid, headers, params, form_def = self._prep_form_post(
+            definition=definition,
             ignore_warnings=ignore_warnings,
             form_id=form_id,
             project_id=project_id,
         )
-
-        with open(file_path, "rb") if file_path is not None else nullcontext() as fd:
-            response = self.session.response_or_error(
-                method="POST",
-                url=self.session.urlformat(self.urls.post, project_id=pid, form_id=fid),
-                logger=log,
-                headers=headers,
-                params=params,
-                data=fd,
-            )
-
+        response = self.session.response_or_error(
+            method="POST",
+            url=self.session.urlformat(self.urls.post, project_id=pid, form_id=fid),
+            logger=log,
+            headers=headers,
+            params=params,
+            data=form_def,
+        )
         data = response.json()
         return data["success"]
 

pyodk/_endpoints/forms.py

@@ -1,6 +1,7 @@
 import logging
 from collections.abc import Callable, Iterable
 from datetime import datetime
+from os import PathLike
 from typing import Any
 
 from pyodk._endpoints import bases
@@ -34,8 +35,8 @@ class URLs(bases.Model):
     class Config:
         frozen = True
 
-    list: str = "projects/{project_id}/forms"
-    get: str = "projects/{project_id}/forms/{form_id}"
+    forms: str = "projects/{project_id}/forms"
+    get: str = f"{forms}/{{form_id}}"
 
 
 class FormService(bases.Service):
@@ -86,7 +87,7 @@ def list(self, project_id: int | None = None) -> list[Form]:
         else:
             response = self.session.response_or_error(
                 method="GET",
-                url=self.session.urlformat(self.urls.list, project_id=pid),
+                url=self.session.urlformat(self.urls.forms, project_id=pid),
                 logger=log,
             )
             data = response.json()
@@ -120,12 +121,48 @@ def get(
             data = response.json()
             return Form(**data)
 
+    def create(
+        self,
+        definition: PathLike | str | bytes,
+        ignore_warnings: bool | None = True,
+        form_id: str | None = None,
+        project_id: int | None = None,
+    ) -> Form:
+        """
+        Create a form.
+
+        :param definition: The path to the file to upload (string or PathLike), or the
+          form definition in memory (string (XML) or bytes (XLS/XLSX)).
+        :param ignore_warnings: If True, create the form if there are XLSForm warnings.
+        :param form_id: The xmlFormId of the Form being referenced.
+        :param project_id: The id of the project this form belongs to.
+        :return: An object representation of the Form's metadata.
+        """
+        fd = FormDraftService(session=self.session, **self._default_kw())
+        pid, fid, headers, params, form_def = fd._prep_form_post(
+            definition=definition,
+            ignore_warnings=ignore_warnings,
+            form_id=form_id,
+            project_id=project_id,
+        )
+        params["publish"] = True
+        response = self.session.response_or_error(
+            method="POST",
+            url=self.session.urlformat(self.urls.forms, project_id=pid),
+            logger=log,
+            headers=headers,
+            params=params,
+            data=form_def,
+        )
+        data = response.json()
+        return Form(**data)
+
     def update(
         self,
         form_id: str,
         project_id: int | None = None,
-        definition: str | None = None,
-        attachments: Iterable[str] | None = None,
+        definition: PathLike | str | bytes | None = None,
+        attachments: Iterable[PathLike | str] | None = None,
         version_updater: Callable[[str], str] | None = None,
     ) -> None:
         """
@@ -151,7 +188,8 @@ def update(
 
         :param form_id: The xmlFormId of the Form being referenced.
         :param project_id: The id of the project this form belongs to.
-        :param definition: The path to a form definition file to upload. The form
+        :param definition: The path to the file to upload (string or PathLike), or the
+          form definition in memory (string (XML) or bytes (XLS/XLSX)). The form
           definition must include an updated version string.
         :param attachments: The paths of the form attachment file(s) to upload.
         :param version_updater: A function that accepts a version name string and returns
@@ -167,7 +205,7 @@ def update(
         # Start a new draft - with a new definition, if provided.
         fp_ids = {"form_id": form_id, "project_id": project_id}
         fd = FormDraftService(session=self.session, **self._default_kw())
-        if not fd.create(file_path=definition, **fp_ids):
+        if not fd.create(definition=definition, **fp_ids):
             raise PyODKError("Form update (form draft create) failed.")
 
         # Upload the attachments, if any.

pyodk/_utils/session.py

@@ -2,6 +2,7 @@
 from string import Formatter
 from typing import Any
 from urllib.parse import quote, urljoin
+from uuid import uuid4
 
 from requests import PreparedRequest, Response
 from requests import Session as RequestsSession
@@ -159,3 +160,10 @@ def response_or_error(
             raise err from e
         else:
             return response
+
+    @staticmethod
+    def get_xform_uuid() -> str:
+        """
+        Get XForm UUID, which is "uuid:" followed by a random uuid v4.
+        """
+        return f"uuid:{uuid4()}"

pyodk/_utils/validators.py

@@ -1,8 +1,10 @@
 from collections.abc import Callable
+from os import PathLike
 from pathlib import Path
 from typing import Any
 
 from pydantic.v1 import validators as v
+from pydantic.v1.errors import PydanticValueError
 from pydantic_core._pydantic_core import ValidationError
 
 from pyodk._utils.utils import coalesce
@@ -20,7 +22,7 @@ def wrap_error(validator: Callable, key: str, value: Any) -> Any:
     """
     try:
         return validator(value)
-    except ValidationError as err:
+    except (ValidationError, PydanticValueError) as err:
         msg = f"{key}: {err!s}"
         raise PyODKError(msg) from err
 
@@ -97,7 +99,7 @@ def validate_dict(*args: dict, key: str) -> int:
     )
 
 
-def validate_file_path(*args: str) -> Path:
+def validate_file_path(*args: PathLike | str) -> Path:
     def validate_fp(f):
         p = v.path_validator(f)
         return v.path_exists_validator(p)