cherry-pick #7032 fix updated_at field in cicd_tasks and cicd_pipelines

backend/python/plugins/azuredevops/azuredevops/streams/builds.py

@@ -25,6 +25,7 @@
 class Builds(Stream):
     tool_model = Build
     domain_types = [DomainType.CICD]
+    domain_models = [devops.CiCDPipelineCommit, devops.CICDPipeline]
 
     def collect(self, state, context) -> Iterable[tuple[object, dict]]:
         repo: GitRepository = context.scope

backend/python/plugins/azuredevops/azuredevops/streams/jobs.py

@@ -28,6 +28,7 @@ class Jobs(Substream):
     tool_model = Job
     domain_types = [DomainType.CICD]
     parent_stream = Builds
+    domain_models = [devops.CICDTask]
 
     def collect(self, state, context, parent: Build) -> Iterable[tuple[object, dict]]:
         repo: GitRepository = context.scope

backend/python/plugins/azuredevops/azuredevops/streams/pull_request_commits.py

@@ -26,6 +26,7 @@ class GitPullRequestCommits(Substream):
     tool_model = GitPullRequestCommit
     domain_types = [DomainType.CODE]
     parent_stream = GitPullRequests
+    domain_models = [code.PullRequestCommit]
 
     def should_run_on(self, scope: GitRepository) -> bool:
         return not scope.is_external()

backend/python/plugins/azuredevops/azuredevops/streams/pull_requests.py

@@ -24,6 +24,7 @@
 class GitPullRequests(Stream):
     tool_model = GitPullRequest
     domain_types = [DomainType.CODE]
+    domain_models = [code.PullRequest]
 
     def should_run_on(self, scope: GitRepository) -> bool:
         return not scope.is_external()

backend/python/pydevlake/pydevlake/model.py

@@ -29,7 +29,6 @@
 from sqlmodel import SQLModel
 from pydevlake import Field
 
-
 inflect_engine = inflect.engine()
 
 
@@ -42,6 +41,10 @@ class Model(SQLModel):
         sa_column=Column(DateTime(), default=datetime.utcnow, onupdate=datetime.utcnow)
     )
 
+    def set_updated_at(self):
+        if self.updated_at is None:
+            self.updated_at = datetime.utcnow()
+
 
 class ToolTable(SQLModel):
     @declared_attr
@@ -132,6 +135,10 @@ class NoPKModel(RawDataOrigin):
         sa_column=Column(DateTime(), default=datetime.utcnow, onupdate=datetime.utcnow)
     )
 
+    def set_updated_at(self):
+        if self.updated_at is None:
+            self.updated_at = datetime.utcnow()
+
 
 class ToolModel(ToolTable, NoPKModel):
     connection_id: Optional[int] = Field(primary_key=True, auto_increment=False)

backend/python/pydevlake/pydevlake/stream.py

@@ -14,7 +14,7 @@
 # limitations under the License.
 
 
-from typing import Iterable, Type
+from typing import Iterable, Type, Optional
 from abc import abstractmethod
 
 from pydevlake.context import Context
@@ -52,6 +52,11 @@ def tool_model(self) -> Type[ToolModel]:
     def domain_types(self) -> list[DomainType]:
         pass
 
+    @property
+    @abstractmethod
+    def domain_models(self) -> Optional[list[Type[DomainModel]]]:
+        pass
+
     def raw_model(self, session) -> Type[RawModel]:
         if self._raw_model is not None:
             return self._raw_model

backend/python/pydevlake/pydevlake/subtasks.py

@@ -179,6 +179,7 @@ def process(self, raw: RawModel, session: Session, ctx: Context):
         tool_model = self.stream.extract(json.loads(raw.data))
         tool_model.set_raw_origin(raw)
         tool_model.connection_id = ctx.connection.id
+        tool_model.set_updated_at()
         session.merge(tool_model)
 
     def delete(self, session, ctx):
@@ -206,9 +207,13 @@ def process(self, tool_model: ToolModel, session: Session, ctx: Context):
 
     def _save(self, tool_model: ToolModel, domain_model: DomainModel, session: Session, connection_id: int):
         domain_model.set_tool_origin(tool_model)
+        domain_model.set_updated_at()
         if isinstance(domain_model, DomainModel):
             domain_model.id = tool_model.domain_id()
         session.merge(domain_model)
 
     def delete(self, session, ctx):
-        pass
+        domain_models = self.stream.domain_models
+        if domain_models is not None:
+            for domain_model in domain_models:
+                session.execute(sql.delete(domain_model).where(domain_model.raw_data_params == self._params(ctx)))

backend/python/pydevlake/pydevlake/testing/testing.py

@@ -78,7 +78,11 @@ def assert_stream_convert(plugin: Union[Plugin, Type[Plugin]], stream_name: str,
     if not isinstance(domain_models, (Iterable, Generator)):
         domain_models = [domain_models]
     for res, exp in zip(domain_models, expected):
-        assert res == exp
+        if (hasattr(res, "__getitem__") and res[0] == "updated_at") or (
+                hasattr(exp, "__getitem__") and exp[0] == "updated_at"):
+            pass
+        else:
+            assert res == exp
 
 
 def assert_stream_run(stream: Stream, connection: Connection, scope: ToolScope, scope_config: ScopeConfig):