Skip to content

Commit

Permalink
add: client.forms.create and update accepts PathLike or bytes
Browse files Browse the repository at this point in the history
- handy if a user already has the file into memory (e.g. because it
  was received in something like pyxform-http, or a test case, or
  because it was created in memory) which saves writing it to a file.
- not implemented for attachments but could do the same there.
- added unit tests for each case of get_definition_data which
  does most of the heavy lifting.
  • Loading branch information
lindsay-stevens committed Apr 18, 2024
1 parent 42b2a1e commit 36fe156
Show file tree
Hide file tree
Showing 8 changed files with 271 additions and 89 deletions.
6 changes: 3 additions & 3 deletions pyodk/_endpoints/form_draft_attachments.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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:
Expand Down
142 changes: 99 additions & 43 deletions pyodk/_endpoints/form_drafts.py
Original file line number Diff line number Diff line change
@@ -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):
Expand Down Expand Up @@ -36,15 +107,16 @@ 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.
Expand All @@ -54,76 +126,60 @@ def _prep_form_post(
pid = pv.validate_project_id(project_id, self.default_project_id)
headers = {}
params = {}
definition_data = None
file_path_stem = None
if file_path is not None:
file_path = Path(pv.validate_file_path(file_path))
file_path_stem = file_path.stem
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 file_path is not None:
if definition is not None:
if ignore_warnings is not None:
key = "ignore_warnings"
params["ignoreWarnings"] = pv.validate_bool(ignore_warnings, key=key)
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(fid),
}
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"]

Expand Down
36 changes: 19 additions & 17 deletions pyodk/_endpoints/forms.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -122,46 +123,46 @@ def get(

def create(
self,
definition: str,
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 a form definition 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 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 = fd._prep_form_post(
file_path=definition,
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
with open(definition, "rb") as fd:
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=fd,
)
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:
"""
Expand All @@ -187,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
Expand All @@ -203,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.
Expand Down
6 changes: 4 additions & 2 deletions pyodk/_utils/validators.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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

Expand Down Expand Up @@ -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)
Expand Down
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,8 @@ dependencies = [
# Install with `pip install pyodk[dev]`.
dev = [
"ruff==0.3.4", # Format and lint
"openpyxl==3.1.2" # Create test XLSX files
"openpyxl==3.1.2", # Create test XLSX files
"xlwt==1.3.0", # Create test XLS files
]
docs = [
"mkdocs==1.5.3",
Expand Down
Loading

0 comments on commit 36fe156

Please sign in to comment.