Add text list attribute for sharepoint integration tags (#179)

* datatypes attribute text list

* payload selector

* IntegrationSharepointPropertySync

* schema

* get update sync properties

* Syntax fix

* import fix

* proeprt sync state

* chore: add extension TODO comment

* get by running ids

* perf: move sharepoint property sync into cognition objects

* perf: move create function into cognition objects

* perf: SYNC_SHAREPOINT_FILE_PROPERTIES task

* perf: add sync_finished for task-master

* perf: remove extra task type

* perf: remove extra task type

* perf: remove unused FINISHED_STATES

---------

Co-authored-by: Lina <lina.lumburovska@kern.ai>
Co-authored-by: andhreljaKern <andrea.hrelja@kern.ai>

Author: 3 people

business_objects/embedding.py

@@ -321,9 +321,13 @@ def __build_payload_selector(
                 data_type != enums.DataTypes.TEXT.value
                 and data_type != enums.DataTypes.LLM_RESPONSE.value
                 and data_type != enums.DataTypes.PERMISSION.value
+                and data_type != enums.DataTypes.TEXT_LIST.value
             ):
                 payload_selector += f"'{attr}', (r.\"data\"->>'{attr}')::{data_type}"
-            elif data_type == enums.DataTypes.PERMISSION.value:
+            elif (
+                data_type == enums.DataTypes.PERMISSION.value
+                or data_type == enums.DataTypes.TEXT_LIST.value
+            ):
                 payload_selector += f"'{attr}', r.\"data\"->'{attr}'"
             else:
                 payload_selector += f"'{attr}', r.\"data\"->>'{attr}'"

business_objects/record.py

@@ -943,7 +943,7 @@ def get_first_no_text_column(project_id: str, record_id: str) -> str:
     (
         SELECT a.name
         FROM attribute a 
-        WHERE data_type NOT IN('{enums.DataTypes.TEXT.value}' , '{enums.DataTypes.CATEGORY.value}', '{enums.DataTypes.LLM_RESPONSE.value}')
+        WHERE data_type NOT IN('{enums.DataTypes.TEXT.value}', '{enums.DataTypes.CATEGORY.value}', '{enums.DataTypes.LLM_RESPONSE.value}', '{enums.DataTypes.TEXT_LIST.value}')
             AND a.state IN ('{enums.AttributeState.AUTOMATICALLY_CREATED.value}','{enums.AttributeState.UPLOADED.value}','{enums.AttributeState.USABLE.value}')
             AND a.project_id = '{project_id}'
         ORDER BY a.relative_position
@@ -968,3 +968,16 @@ def get_record_ids_by_running_ids(project_id: str, running_ids: List[int]) -> Li
             .all()
         )
     ]
+
+
+def get_records_by_running_ids(project_id: str, running_ids: List[int]) -> List[str]:
+    return (
+        session.query(Record)
+        .filter(
+            Record.project_id == project_id,
+            Record.data[attribute.get_running_id_name(project_id)]
+            .as_integer()
+            .in_(running_ids),
+        )
+        .all()
+    )

cognition_objects/integration_sharepoint_property_sync.py

@@ -0,0 +1,49 @@
+from typing import List, Optional, Dict, Any
+from sqlalchemy.orm.attributes import flag_modified
+from ..models import IntegrationSharepointPropertySync
+from ..enums import SharepointPropertySyncState
+from ..business_objects import general
+from ..session import session
+
+
+def get_by_integration_id(
+    integration_id: str,
+) -> IntegrationSharepointPropertySync:
+    return (
+        session.query(IntegrationSharepointPropertySync)
+        .filter(IntegrationSharepointPropertySync.integration_id == integration_id)
+        .first()
+    )
+
+
+def create(
+    integration_id: str,
+    config: Optional[Dict[str, Any]] = None,
+    logs: Optional[List[Dict[str, Any]]] = None,
+    with_commit: bool = True,
+) -> IntegrationSharepointPropertySync:
+    integration_sync = IntegrationSharepointPropertySync(
+        integration_id=integration_id,
+        config=config or {},
+        logs=logs or [],
+    )
+    session.add(integration_sync)
+    general.flush_or_commit(with_commit)
+    return integration_sync
+
+
+def update(
+    integration_id: str,
+    config: Optional[Dict[str, Any]] = None,
+    logs: Optional[List[Dict[str, Any]]] = None,
+    with_commit: bool = True,
+) -> IntegrationSharepointPropertySync:
+    integration_sync = get_by_integration_id(integration_id)
+    if config is not None:
+        integration_sync.config = config
+        flag_modified(integration_sync, "config")
+    if logs is not None:
+        integration_sync.logs = logs
+        flag_modified(integration_sync, "logs")
+    general.flush_or_commit(with_commit)
+    return integration_sync

enums.py

@@ -11,6 +11,7 @@ class DataTypes(Enum):
     LLM_RESPONSE = "LLM_RESPONSE"
     EMBEDDING_LIST = "EMBEDDING_LIST"  # only for embeddings & default hidden
     PERMISSION = "PERMISSION"  # used for access control
+    TEXT_LIST = "TEXT_LIST"
     UNKNOWN = "UNKNOWN"
 
 
@@ -168,6 +169,7 @@ class Tablenames(Enum):
     INTEGRATION_PDF = "pdf"
     INTEGRATION_SHAREPOINT = "sharepoint"
     STEP_TEMPLATES = "step_templates"  # templates for strategy steps
+    INTEGRATION_SHAREPOINT_PROPERTY_SYNC = "sharepoint_property_sync"
     CONVERSATION_TAG = "conversation_tag"  # config of tags used in conversations
     CONVERSATION_TAG_ASSOCIATION = (
         "conversation_tag_association"  # association between conversation and tags
@@ -946,3 +948,10 @@ def from_string(value: str):
             raise KeyError(
                 f"Could not parse CognitionIntegrationType from string '{changed_value}'"
             )
+
+
+class SharepointPropertySyncState(Enum):
+    CREATED = "CREATED"
+    RUNNING = "RUNNING"
+    COMPLETED = "COMPLETED"
+    FAILED = "FAILED"

integration_objects/manager.py

@@ -92,6 +92,8 @@ def get_existing_integration_records(
     integration_id: str,
     by: str = "source",
 ) -> Dict[str, object]:
+    # TODO(extension): make return type Dict[str, List[object]]
+    # once an object_id can reference multiple different integration records
     return {
         getattr(record, by, record.source): record
         for record in get_all_by_integration_id(IntegrationModel, integration_id)

models.py

@@ -2390,6 +2390,27 @@ class IntegrationSharepoint(Base):
     file_properties = Column(JSON)
 
 
+class IntegrationSharepointPropertySync(Base):
+    __tablename__ = Tablenames.INTEGRATION_SHAREPOINT_PROPERTY_SYNC.value
+    __table_args__ = (UniqueConstraint("integration_id"), {"schema": "integration"})
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    created_by = Column(
+        UUID(as_uuid=True),
+        ForeignKey(f"{Tablenames.USER.value}.id", ondelete="SET NULL"),
+        index=True,
+    )
+    created_at = Column(DateTime, default=sql.func.now())
+    updated_at = Column(DateTime, onupdate=sql.func.now())
+    integration_id = Column(
+        UUID(as_uuid=True),
+        ForeignKey(f"cognition.{Tablenames.INTEGRATION.value}.id", ondelete="CASCADE"),
+        index=True,
+    )
+    config = Column(JSON)  # JSON object containing the rules for property sync
+    logs = Column(ARRAY(String))
+    state = Column(String)
+
+
 class CognitionConversationTag(Base):
     __tablename__ = Tablenames.CONVERSATION_TAG.value
     __table_args__ = {"schema": "cognition"}