Merge pull request #330 from mustafa3rsan/add-states-sourcetable-for-linear

Add states sourcetable for linear

Author: sabrikaragonen

ingestr/src/linear/__init__.py

@@ -3,7 +3,7 @@
 import dlt
 import pendulum
 
-from .helpers import _normalize_issue, _normalize_team, _paginate
+from .helpers import _paginate, normalize_dictionaries
 
 
 def _get_date_range(updated_at, start_date):
@@ -99,7 +99,25 @@ def _get_date_range(updated_at, start_date):
   }
 }
 """
-
+WORKFLOW_STATES_QUERY = """
+query WorkflowStates($cursor: String) {
+  workflowStates(first: 50, after: $cursor) {
+    nodes { 
+      archivedAt
+      color
+      createdAt
+      id
+      inheritedFrom { id }
+      name
+      position
+      team { id  }
+      type
+      updatedAt
+    }
+    pageInfo { hasNextPage endCursor }
+  }
+}
+"""
 
 @dlt.source(name="linear", max_table_nesting=0)
 def linear_source(
@@ -122,7 +140,7 @@ def issues(
         for item in _paginate(api_key, ISSUES_QUERY, "issues"):
             if pendulum.parse(item["updatedAt"]) >= current_start_date:
                 if pendulum.parse(item["updatedAt"]) <= current_end_date:
-                    yield _normalize_issue(item)
+                    yield normalize_dictionaries(item)
 
     @dlt.resource(name="projects", primary_key="id", write_disposition="merge")
     def projects(
@@ -139,7 +157,7 @@ def projects(
         for item in _paginate(api_key, PROJECTS_QUERY, "projects"):
             if pendulum.parse(item["updatedAt"]) >= current_start_date:
                 if pendulum.parse(item["updatedAt"]) <= current_end_date:
-                    yield item
+                    yield normalize_dictionaries(item)
 
     @dlt.resource(name="teams", primary_key="id", write_disposition="merge")
     def teams(
@@ -158,7 +176,7 @@ def teams(
         for item in _paginate(api_key, TEAMS_QUERY, "teams"):
             if pendulum.parse(item["updatedAt"]) >= current_start_date:
                 if pendulum.parse(item["updatedAt"]) <= current_end_date:
-                    yield _normalize_team(item)
+                    yield normalize_dictionaries(item)
 
     @dlt.resource(name="users", primary_key="id", write_disposition="merge")
     def users(
@@ -175,6 +193,23 @@ def users(
         for item in _paginate(api_key, USERS_QUERY, "users"):
             if pendulum.parse(item["updatedAt"]) >= current_start_date:
                 if pendulum.parse(item["updatedAt"]) <= current_end_date:
-                    yield item
+                    yield normalize_dictionaries(item)
+
+    @dlt.resource(name="workflow_states", primary_key="id", write_disposition="merge")
+    def workflow_states(
+        updated_at: dlt.sources.incremental[str] = dlt.sources.incremental(
+            "updatedAt",
+            initial_value=start_date.isoformat(),
+            end_value=end_date.isoformat() if end_date else None,
+            range_start="closed",
+            range_end="closed",
+        ),
+    ) -> Iterator[Dict[str, Any]]:
+        current_start_date, current_end_date = _get_date_range(updated_at, start_date)
+
+        for item in _paginate(api_key, WORKFLOW_STATES_QUERY, "workflowStates"):
+            if pendulum.parse(item["updatedAt"]) >= current_start_date:
+                if pendulum.parse(item["updatedAt"]) <= current_end_date:
+                    yield normalize_dictionaries(item)
+    return [issues, projects, teams, users, workflow_states]
 
-    return issues, projects, teams, users

ingestr/src/linear/helpers.py

@@ -32,41 +32,24 @@ def _paginate(api_key: str, query: str, root: str) -> Iterator[Dict[str, Any]]:
         cursor = data["pageInfo"]["endCursor"]
 
 
-def _normalize_issue(item: Dict[str, Any]) -> Dict[str, Any]:
-    field_mapping = {
-        "assignee": "assignee_id",
-        "creator": "creator_id",
-        "state": "state_id",
-        "cycle": "cycle_id",
-        "project": "project_id",
-    }
-    for key, value in field_mapping.items():
-        if item.get(key):
-            item[value] = item[key]["id"]
-            del item[key]
-        else:
-            item[value] = None
-            del item[key]
-    json_fields = [
-        "comments",
-        "subscribers",
-        "attachments",
-        "labels",
-        "subtasks",
-        "projects",
-        "memberships",
-        "members",
-    ]
-    for field in json_fields:
-        if item.get(field):
-            item[f"{field}"] = item[field].get("nodes", [])
 
-    return item
 
-
-def _normalize_team(item: Dict[str, Any]) -> Dict[str, Any]:
-    json_fields = ["memberships", "members", "projects"]
-    for field in json_fields:
-        if item.get(field):
-            item[f"{field}"] = item[field].get("nodes", [])
-    return item
+def normalize_dictionaries(item: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Automatically normalize dictionary fields by detecting their structure:
+    - Convert nested objects with 'id' field to {field_name}_id
+    - Convert objects with 'nodes' field to arrays
+    """
+    normalized_item = item.copy()
+    
+    for key, value in list(normalized_item.items()):
+        if isinstance(value, dict):
+            # If the dict has an 'id' field, replace with {key}_id
+            if 'id' in value:
+                normalized_item[f"{key}_id"] = value['id']
+                del normalized_item[key]
+            # If the dict has 'nodes' field, extract the nodes array
+            elif 'nodes' in value:
+                normalized_item[key] = value['nodes']
+    
+    return normalized_item

ingestr/src/sources.py

@@ -3160,7 +3160,7 @@ def dlt_source(self, uri: str, table: str, **kwargs):
         if api_key is None:
             raise MissingValueError("api_key", "Linear")
 
-        if table not in ["issues", "projects", "teams", "users"]:
+        if table not in ["issues", "projects", "teams", "users", "workflow_states"]:
             raise UnsupportedResourceError(table, "Linear")
 
         start_date = kwargs.get("interval_start")

requirements.txt

@@ -214,6 +214,8 @@ googleapis-common-protos==1.69.0
     #   google-api-core
     #   grpc-google-iam-v1
     #   grpcio-status
+greenlet==3.2.3
+    # via sqlalchemy
 grpc-google-iam-v1==0.14.2
     # via google-cloud-spanner
 grpc-interceptor==0.15.4