feat: gerrit python plugin (#6786)

* feat: gerrit python plugin

Signed-off-by: Ji Bin <matrixji@live.com>

* feat: gerrit plugin support incremental sync

Signed-off-by: Ji Bin <matrixji@live.com>

---------

Signed-off-by: Ji Bin <matrixji@live.com>
Co-authored-by: Lynwee <linwei.hou@merico.dev>

backend/python/plugins/gerrit/README.md

@@ -0,0 +1,18 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+
+# Gerrit Python Plugin

backend/python/plugins/gerrit/build.sh

@@ -0,0 +1,20 @@
+#!/bin/sh
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+cd "$(dirname "$0")"
+poetry install

backend/python/plugins/gerrit/gerrit/__init__.py

@@ -0,0 +1,14 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

backend/python/plugins/gerrit/gerrit/api.py

@@ -0,0 +1,120 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from base64 import b64encode
+from os import environ
+from typing import Optional
+from urllib.parse import urlparse
+from datetime import datetime, timedelta
+from MySQLdb import connect as mysql_connect, Error as MySQLError
+from pydevlake.api import API, Request, Response, request_hook, response_hook, Paginator
+from gerrit.models import GerritChange
+
+
+# TODO: implement pagination
+class GerritPaginator(Paginator):
+    def get_items(self, response) -> Optional[list[object]]:
+        return response.json
+
+    def get_next_page_id(self, response) -> Optional[str]:
+        return []
+
+    def set_next_page_param(self, request, next_page_id):
+        pass
+
+
+class GerritApi(API):
+    # paginator = GerritPaginator()
+
+    def __init__(self, connection=None):
+        super().__init__(connection)
+        self.db_conn = None
+
+    def auto_connect(self):
+        if self.db_conn:
+            try:
+                self.db_conn.ping()
+                return
+            except MySQLError as e:
+                self.db_conn.close()
+        self.db_conn = None
+        if 'DB_URL' in environ:
+            parsed_url = urlparse((environ['DB_URL']))
+            connection_args = {
+                'user': parsed_url.username,
+                'password': parsed_url.password,
+                'host': parsed_url.hostname,
+                'port': parsed_url.port or 3306,  # Default MySQL port
+                # Remove leading slash from path
+                'database': parsed_url.path[1:]
+            }
+            try:
+                self.db_conn = mysql_connect(**connection_args)
+            except MySQLError as e:
+                print(f"Error connecting to MySQL: {e}")
+
+    @property
+    def base_url(self):
+        return self.connection.url
+
+    @request_hook
+    def authenticate(self, request: Request):
+        conn = self.connection
+        if conn.username and conn.password:
+            user_pass = f"{conn.username}:{conn.password.get_secret_value()}".encode()
+            basic_auth = b64encode(user_pass).decode()
+            request.headers["Authorization"] = f"Basic {basic_auth}"
+
+    @response_hook
+    def remove_extra_content_in_json(self, response: Response):
+        # remove ")]}'"
+        if response.body.startswith(b")]}'"):
+            response.body = response.body[4:]
+
+    def my_profile(self):
+        return self.get("accounts/self")
+
+    def projects(self):
+        # TODO: use pagination
+        projects_uri = "projects/?type=CODE&n=10000"
+        if self.connection.pattern:
+            projects_uri += f"&r={self.connection.pattern}"
+        return self.get(projects_uri)
+
+    def changes(self, project_name: str):
+        # TODO: use pagination
+        self.auto_connect()
+        start_date = None
+        if self.db_conn:
+            cursor = self.db_conn.cursor()
+            try:
+                cursor.execute(
+                    f"SELECT updated_at FROM _tool_gerrit_gerritchanges WHERE id like '{project_name}~%' ORDER BY updated_at desc limit 1")
+                last_updated = cursor.fetchone()
+                if last_updated and len(last_updated) > 0:
+                    last_updated = last_updated[0] - timedelta(days=1)
+                    start_date = datetime.strftime(last_updated, "%Y-%m-%d")
+            except MySQLError as e:
+                print(f"Error fetching last updated date: {e}")
+            cursor.close()
+        if start_date:
+            return self.get(f"changes/?q=p:{project_name}+after:{start_date}&o=CURRENT_REVISION&o=ALL_COMMITS&o=DETAILED_ACCOUNTS&no-limit")
+        return self.get(f"changes/?q=p:{project_name}&o=CURRENT_REVISION&o=ALL_COMMITS&o=DETAILED_ACCOUNTS&no-limit")
+
+    def change_detail(self, change_id: str):
+        return self.get(f"changes/{change_id}/detail")
+
+    def account(self, account_id: int):
+        return self.get(f"accounts/{account_id}")

backend/python/plugins/gerrit/gerrit/main.py

@@ -0,0 +1,117 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+from urllib.parse import urlparse
+from gerrit.streams.changes import GerritChanges
+from gerrit.streams.change_commits import GerritChangeCommits
+from gerrit.api import GerritApi
+from gerrit.models import GerritConnection, GerritProject, GerritProjectConfig
+
+from pydevlake.api import APIException
+from pydevlake.domain_layer.code import Repo
+from pydevlake.message import (
+    PipelineTask,
+    RemoteScopeGroup,
+    TestConnectionResult,
+)
+from pydevlake.model import (
+    Connection,
+    DomainType,
+    ScopeConfig,
+)
+from pydevlake.pipeline_tasks import gitextractor, refdiff
+from pydevlake.plugin import Plugin
+from pydevlake.stream import Stream
+
+
+logger = logging.getLogger()
+
+
+class GerritPlugin(Plugin):
+    @property
+    def connection_type(self):
+        return GerritConnection
+
+    @property
+    def tool_scope_type(self):
+        return GerritProject
+
+    @property
+    def scope_config_type(self):
+        return GerritProjectConfig
+
+    def domain_scopes(self, gerrit_project: GerritProject):
+        yield Repo(
+            name=gerrit_project.name,
+            url=gerrit_project.url,
+        )
+
+    def remote_scope_groups(self, connection: Connection) -> list[RemoteScopeGroup]:
+        yield RemoteScopeGroup(
+            id=f"{connection.id}:default",
+            name="Code Repositories",
+        )
+
+    def remote_scopes(self, connection: Connection, group_id: str) -> list[GerritProject]:
+        api = GerritApi(connection)
+        json_data = api.projects().json
+        for project_name in json_data:
+            yield GerritProject(
+                id=project_name,
+                name=project_name,
+                url=connection.url + project_name,
+            )
+
+    def test_connection(self, connection: Connection):
+        api = GerritApi(connection)
+        message = None
+        try:
+            res = api.projects()
+        except APIException as e:
+            res = e.response
+            message = "HTTP Error: " + str(res.status)
+        return TestConnectionResult.from_api_response(res, message)
+
+    def extra_tasks(
+        self, scope: GerritProject, config: ScopeConfig, connection: GerritConnection
+    ) -> list[PipelineTask]:
+        if DomainType.CODE in config.domain_types:
+            url = urlparse(scope.url)
+            if connection.username and connection.password:
+                url = url._replace(
+                    netloc=f"{connection.username}:{connection.password.get_secret_value()}@{url.netloc}"
+                )
+            yield gitextractor(url.geturl(), scope.name, scope.domain_id(), connection.proxy)
+
+    def extra_stages(
+        self,
+        scope_config_pairs: list[tuple[GerritProject, ScopeConfig]],
+        connection: GerritConnection,
+    ) -> list[list[PipelineTask]]:
+        for scope, config in scope_config_pairs:
+            if DomainType.CODE in config.domain_types:
+                yield [refdiff(scope.id, config.refdiff)]
+
+    @property
+    def streams(self) -> list[Stream]:
+        return [
+            GerritChanges,
+            GerritChangeCommits,
+        ]
+
+
+if __name__ == "__main__":
+    GerritPlugin.start()

backend/python/plugins/gerrit/gerrit/migrations.py

@@ -0,0 +1,70 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from typing import Optional
+from datetime import datetime
+
+from pydantic import SecretStr
+
+from pydevlake import ToolModel, Connection, Field
+from pydevlake.migration import migration, MigrationScriptBuilder
+from pydevlake.model import ScopeConfig, ToolScope
+from pydevlake.pipeline_tasks import RefDiffOptions
+
+
+@migration(20240108000001, name="initialize schemas for gerrit")
+def init_schemas(b: MigrationScriptBuilder):
+    class GerritConnection(Connection):
+        endpoint: str
+        username: Optional[str]
+        password: Optional[SecretStr]
+        pattern: Optional[str]
+
+    class GerritProject(ToolScope):
+        name: str
+        url: str
+
+    class GerritProjectConfig(ScopeConfig):
+        refdiff: Optional[RefDiffOptions]
+
+    class GerritChange(ToolModel):
+        id: str = Field(primary_key=True)
+        change_id: str
+        change_number: int
+        subject: str
+        status: str
+        branch: str
+        created_date: datetime
+        merged_date: Optional[datetime]
+        closed_date: Optional[datetime]
+        current_revision: Optional[str]
+        owner_name: Optional[str]
+        owner_email: Optional[str]
+        revisions_json: Optional[str]
+
+    class GerritChangeCommit(ToolModel):
+        commit_id: str = Field(primary_key=True)
+        pull_request_id: str
+        author_name: str
+        author_email: str
+        author_date: datetime
+
+    b.create_tables(
+        GerritConnection,
+        GerritProject,
+        GerritProjectConfig,
+        GerritChange,
+        GerritChangeCommit,
+    )