feat: add pydantic models for projects endpoint

.isort.cfg

@@ -1,4 +1,4 @@
 [settings]
 profile = black
 combine_as_imports = true
-known_third_party = backoff,cachetools,requests,requests_mock,typing_extensions
+known_third_party = backoff,cachetools,pydantic,pytest,requests,requests_mock,typing_extensions

VERSION

@@ -1 +1 @@
-2.0.1
+2.1.0

caplena/errors.py

@@ -0,0 +1,18 @@
+from typing import Any, Dict
+
+
+class DuplicatedTopicsError(Exception):
+    """Exception that is thrown when there are two or more topics with identical names
+    provided for TTACell.
+
+    :param message: A brief human-readable message providing more details about the error
+        that has occurred. Please note that error messages might change and are therefore
+        not suitable for programmatic error handling.
+    :param duplicates: A dict which specifies which topic ids had duplicates.
+        Keys are topic ids and values are number of occurences.
+    """
+
+    def __init__(self, message: str, duplicates: Dict[str, int], *args: Any) -> None:
+        self.message = message
+        self.duplicates = duplicates
+        super().__init__(message, *args)

caplena/models/projects.py

@@ -0,0 +1,182 @@
+from datetime import datetime
+from enum import Enum
+from typing import Any, Callable, ClassVar, Dict, List, Optional, Sequence, Union, cast
+
+import pydantic
+
+from caplena.validations.projects import validate_no_duplicated_topics
+
+
+class NonTTAColumnType(Enum):
+    numerical = "numerical"
+    date = "date"
+    boolean = "boolean"
+    text = "text"
+
+
+class TTAColumnType(Enum):
+    text_to_analyze = "text_to_analyze"
+
+
+class NonTTAColumnDefinition(pydantic.BaseModel):
+    """Column definition for NonTextToAnalyze column."""
+
+    ref: str
+    type: NonTTAColumnType
+    name: str
+    convertor: ClassVar[Dict[str, Callable[[str], Union[int, datetime, bool, str]]]] = {
+        NonTTAColumnType.numerical.value: int,
+        NonTTAColumnType.date.value: datetime.fromisoformat,
+        NonTTAColumnType.boolean.value: bool,
+        NonTTAColumnType.text.value: str,
+    }
+
+    # we want the enum value when serialising to .dict()
+    model_config = pydantic.ConfigDict(use_enum_values=True)
+
+    def convert_to_type(self, val: str) -> Union[int, datetime, bool, str]:
+        # we use the enum's values here since the class is configured to always use enum's values
+        type_val = cast(str, self.type)  # Config.use_enum_values converts type to .value (str)
+        return self.convertor[type_val](val)
+
+    def build_cell(self, ref: str, value: str) -> "NonTTACell":
+        return NonTTACell(ref=ref, value=self.convert_to_type(value))
+
+
+class Sentiment(pydantic.BaseModel):
+    code: Optional[int] = None
+    label: Optional[str] = None
+
+
+class TopicSentiment(Enum):
+    NEUTRAL = "neutral"
+    POSITIVE = "positive"
+    NEGATIVE = "negative"
+    ANY = "any"
+
+
+class TopicDefinition(pydantic.BaseModel):
+    label: str
+    sentiment_enabled: bool
+    category: str
+    color: Optional[str] = None
+    description: Optional[str] = None
+    sentiment_neutral: Optional[Sentiment] = None
+    sentiment_negative: Optional[Sentiment] = None
+    sentiment_positive: Optional[Sentiment] = None
+
+    def build_topic(self, id: str, sentiment: TopicSentiment = TopicSentiment.ANY) -> "Topic":
+        _sentiment = sentiment if self.sentiment_enabled else TopicSentiment.ANY
+        return Topic(id=id, sentiment=_sentiment)
+
+
+class TTAColumnDefinition(pydantic.BaseModel):
+    """Column definition for TextToAnalyze column."""
+
+    ref: str
+    name: str
+    topics: List[TopicDefinition]
+    type: TTAColumnType = TTAColumnType.text_to_analyze
+    description: Optional[str] = None
+    model_config = pydantic.ConfigDict(use_enum_values=True)
+
+    def convert_to_type(self, val: Any) -> str:
+        return str(val)
+
+    def build_cell(
+        self, ref: str, topics: List["Topic"], value: str, was_reviewed: bool
+    ) -> "TTACell":
+        return TTACell(
+            ref=ref, topics=topics, value=self.convert_to_type(value), was_reviewed=was_reviewed
+        )
+
+
+ColumnDefinition = Union[TTAColumnDefinition, NonTTAColumnDefinition]
+
+
+class Topic(pydantic.BaseModel):
+    id: str
+    sentiment: TopicSentiment
+    model_config = pydantic.ConfigDict(use_enum_values=True)
+
+
+class TTACell(pydantic.BaseModel):
+    """Cell definition for TextToAnalyze cell."""
+
+    ref: str
+    topics: List[Topic]
+    value: str
+    was_reviewed: bool = True
+
+    @pydantic.validator("topics")
+    def topics_shouldnt_have_duplicates(cls, topics: List[Topic]) -> List[Topic]:
+        validate_no_duplicated_topics(topics)
+        return topics
+
+
+class NonTTACell(pydantic.BaseModel):
+    """Cell definition for NonTextToAnalyze cell."""
+
+    ref: str
+    value: Optional[Union[int, str, bool, datetime]] = None
+
+
+Cell = Union[TTACell, NonTTACell]
+
+
+class MultipleCellPayload(pydantic.BaseModel):
+    cells: List[Cell]
+
+
+class RowPayload(pydantic.BaseModel):
+    columns: List[Cell]
+
+
+class MultipleRowPayload(pydantic.BaseModel):
+    rows: List[RowPayload]
+
+
+class ProjectLanguage(Enum):
+    AF = "af"
+    SQ = "sq"
+    EU = "eu"
+    CA = "ca"
+    CS = "cs"
+    DA = "da"
+    NL = "nl"
+    EN = "en"
+    ET = "et"
+    FI = "fi"
+    FR = "fr"
+    GL = "gl"
+    DE = "de"
+    EL = "el"
+    HU = "hu"
+    IS = "is"
+    IT = "it"
+    LB = "lb"
+    LT = "lt"
+    LV = "lv"
+    MK = "mk"
+    NO = "no"
+    PL = "pl"
+    PT = "pt"
+    RO = "ro"
+    SR = "sr"
+    SK = "sk"
+    SL = "sl"
+    ES = "es"
+    SV = "sv"
+    TR = "tr"
+
+
+class ProjectSettings(pydantic.BaseModel):
+    """Projects settings for project creation."""
+
+    name: str
+    language: ProjectLanguage
+    columns: Sequence[ColumnDefinition]
+    tags: List[str] = pydantic.Field(default_factory=list)
+    translation_engine: Optional[str] = None
+    anonymize_pii: Optional[Any] = pydantic.Field(default=None)
+    model_config = pydantic.ConfigDict(use_enum_values=True)

caplena/validations/projects.py

@@ -0,0 +1,26 @@
+from typing import TYPE_CHECKING, Dict, List
+
+from caplena.errors import DuplicatedTopicsError
+
+# avoid circular dependency when not type-checking
+if TYPE_CHECKING:
+    from caplena.models.projects import Topic
+
+
+def validate_no_duplicated_topics(topics: List["Topic"]) -> None:
+    """Validates whether there are duplicate topics in a list provided.
+
+    :param topics: A list of topics to check for duplicates.
+    :raises caplena.errors.DuplicatedTopicsError: An exception raised if there are duplicate topics.
+    """
+    id_to_occurences: Dict[str, List[Topic]] = {}
+    for topic in topics:
+        id_to_occurences.setdefault(topic.id, []).append(topic)
+    duplicates = {id: len(topics) for id, topics in id_to_occurences.items() if len(topics) > 1}
+    if duplicates:
+        raise DuplicatedTopicsError(
+            "A cell cannot contain duplicated topics. The topics with following ids have been"
+            "provided multiple times. `.duplicates` attribute of this exception contains: "
+            f"{duplicates}",
+            duplicates,
+        )

caplena/version.py

@@ -1 +1 @@
-__version__ = "2.0.1"
+__version__ = "2.1.0"

docs/source/conf.py

@@ -22,8 +22,8 @@
 project = "Caplena"
 copyright = "2024, Caplena"
 author = "Caplena"
-version = "2.0.1"
-release = "2.0.1"
+version = "2.1.0"
+release = "2.1.0"
 
 
 # -- General configuration ---------------------------------------------------

docs/source/creating-projects.rst

@@ -17,25 +17,43 @@ Next, we'll build the project's columns which defines the schema of the rows to
 
 .. code-block:: python
 
-  columns = [
-    {
-        "name": "Survey Response ID", # name is what is shown in the User Interface
-        "ref": "id", # ref is a unique identifier for the column in the project
-        "type": "numerical"
-    },
-    {
-        "name": "Why did you give this rating?",
-        "ref": "nps_why",
-        "type": "text_to_analyze",
-        "description": "Please explain the rating in a few sentences."
-    }
+  from caplena.models.projects import (
+      NonTTAColumnDefinition,
+      NonTTAColumnType,
+      TTAColumnDefinition,
+      TTAColumnType,
+   )
+
+  columns=[
+      NonTTAColumnDefinition(
+          ref="id",  # ref is a unique identifier for the column in the project
+          name="Survey Response ID", # name is what is shown in the User Interface
+          type=NonTTAColumnType.numerical,
+      ),
+      TTAColumnDefinition(
+          ref="nps_why",
+          name="Why did you give this rating?",
+          type=TTAColumnType.text_to_analyze,
+          description="Please explain the rating in a few sentences.",
+          topics=[],
+      ),
   ]
 
+
 Now we're ready to create the project:
 
 .. code-block:: python
 
-  new_project = client.projects.create(name="NPS Study", language='en', columns=columns, tags=["NPS"])
+  from caplena.models.projects import ProjectLanguage, ProjectSettings
+
+  project_settings = ProjectSettings(
+      name="NPS Study",
+      language=ProjectLanguage.EN,
+      columns=columns,
+      tags=["NPS"],
+  ).model_dump(exclude_none=True)
+
+  new_project = client.projects.create(**project_settings)
 
 Optionally, we can pass :code:`translation_engine=google_translate` to translate rows automatically using Google Translate.
 
@@ -53,11 +71,25 @@ The ordering of columns within a row does not matter as columns are referenced u
 
 .. code-block:: python
 
+  from caplena.models.projects import (
+     MultipleRowPayload,
+     RowPayload,
+     NonTTACell,
+     TTACell,
+   )
+
   # generate fake rows
-  rows = [
-    {"columns": [{"ref": "id", "value": i}, {"ref":"nps_why", "value": f"Row {i}"}]}
-     for i in range(100)
-  ]
+  rows = MultipleRowPayload(
+      rows=[
+          RowPayload(
+              columns=[
+                  NonTTACell(ref="id", value=i),
+                  TTACell(ref="nps_why", value=f"Row {i}", topics=[]),
+              ]
+          ) for i in range(100)
+      ]
+  ).model_dump()["rows"]
+
   # batch rows, we'll use numpy for this
   import numpy as np
   n_batches = np.ceil(len(rows)/20) # compute the number of batches needed

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "flit_core.buildapi"
 
 [project]
 name = "caplena"
-version = "2.0.1"
+version = "2.1.0"
 authors = [{name = "Caplena", email = "support@caplena.com"}]
 maintainers = [{name = "Pascal de Buren", email = "pascal@caplena.com"}]
 keywords = ["caplena", "api", "nlp", "customer feedback"]
@@ -33,6 +33,7 @@ dependencies = [
     "typing-extensions >=4.0.0",
     "backoff >=2.2.0",
     "cachetools>=5.0.0",
+    "pydantic>=2.0.0",
 ]
 
 [project.optional-dependencies]
@@ -42,6 +43,9 @@ test = [
     "requests-mock",
     "types-cachetools",
     "types-six",
+    "types-Pygments",
+    "types-colorama",
+    "types-setuptools",
 ]
 doc = [
     "sphinx",