Module ogr.services.gitlab

class GitlabIssue(BaseIssue):
    _raw_issue: _GitlabIssue

    @property
    def title(self) -> str:
        return self._raw_issue.title

    @title.setter
    def title(self, new_title: str) -> None:
        self._raw_issue.title = new_title
        self._raw_issue.save()

    @property
    def id(self) -> int:
        return self._raw_issue.iid

    @property
    def private(self) -> bool:
        return self._raw_issue.confidential

    @property
    def status(self) -> IssueStatus:
        return (
            IssueStatus.open
            if self._raw_issue.state == "opened"
            else IssueStatus[self._raw_issue.state]
        )

    @property
    def url(self) -> str:
        return self._raw_issue.web_url

    @property
    def assignees(self) -> list:
        try:
            return self._raw_issue.assignees
        except AttributeError:
            return None  # if issue has no assignees, the attribute is not present

    @property
    def description(self) -> str:
        return self._raw_issue.description

    @description.setter
    def description(self, new_description: str) -> None:
        self._raw_issue.description = new_description
        self._raw_issue.save()

    @property
    def author(self) -> str:
        return self._raw_issue.author["username"]

    @property
    def created(self) -> datetime.datetime:
        return self._raw_issue.created_at

    @property
    def labels(self) -> list[IssueLabel]:
        return [GitlabIssueLabel(label, self) for label in self._raw_issue.labels]

    def __str__(self) -> str:
        return "Gitlab" + super().__str__()

    @staticmethod
    def create(
        project: "ogr_gitlab.GitlabProject",
        title: str,
        body: str,
        private: Optional[bool] = None,
        labels: Optional[list[str]] = None,
        assignees: Optional[list[str]] = None,
    ) -> "Issue":
        if not project.has_issues:
            raise IssueTrackerDisabled()

        assignee_ids = []
        for user in assignees or []:
            users_list = project.service.gitlab_instance.users.list(username=user)

            if not users_list:
                raise GitlabAPIException(f"Unable to find '{user}' username")

            assignee_ids.append(str(users_list[0].id))

        data = {"title": title, "description": body}
        if labels:
            data["labels"] = ",".join(labels)
        if assignees:
            data["assignee_ids"] = ",".join(assignee_ids)

        issue = project.gitlab_repo.issues.create(data, confidential=private)
        return GitlabIssue(issue, project)

    @staticmethod
    def get(project: "ogr_gitlab.GitlabProject", issue_id: int) -> "Issue":
        if not project.has_issues:
            raise IssueTrackerDisabled()

        try:
            return GitlabIssue(project.gitlab_repo.issues.get(issue_id), project)
        except gitlab.exceptions.GitlabGetError as ex:
            raise GitlabAPIException(f"Issue {issue_id} was not found. ") from ex

    @staticmethod
    def get_list(
        project: "ogr_gitlab.GitlabProject",
        status: IssueStatus = IssueStatus.open,
        author: Optional[str] = None,
        assignee: Optional[str] = None,
        labels: Optional[list[str]] = None,
    ) -> list["Issue"]:
        if not project.has_issues:
            raise IssueTrackerDisabled()

        # Gitlab API has status 'opened', not 'open'
        parameters: dict[str, Union[str, list[str], bool]] = {
            "state": status.name if status != IssueStatus.open else "opened",
            "order_by": "updated_at",
            "sort": "desc",
            "all": True,
        }
        if author:
            parameters["author_username"] = author
        if assignee:
            parameters["assignee_username"] = assignee
        if labels:
            parameters["labels"] = labels

        issues = project.gitlab_repo.issues.list(**parameters)
        return [GitlabIssue(issue, project) for issue in issues]

    def _get_all_comments(self) -> list[IssueComment]:
        return [
            GitlabIssueComment(parent=self, raw_comment=raw_comment)
            for raw_comment in self._raw_issue.notes.list(sort="asc", all=True)
        ]

    def comment(self, body: str) -> IssueComment:
        comment = self._raw_issue.notes.create({"body": body})
        return GitlabIssueComment(parent=self, raw_comment=comment)

    def close(self) -> "Issue":
        self._raw_issue.state_event = "close"
        self._raw_issue.save()
        return self

    def add_label(self, *labels: str) -> None:
        for label in labels:
            self._raw_issue.labels.append(label)
        self._raw_issue.save()

    def add_assignee(self, *assignees: str) -> None:
        assignee_ids = self._raw_issue.__dict__.get("assignee_ids") or []
        for assignee in assignees:
            users = self.project.service.gitlab_instance.users.list(  # type: ignore
                username=assignee,
            )
            if not users:
                raise GitlabAPIException(f"Unable to find '{assignee}' username")
            uid = str(users[0].id)
            if uid not in assignee_ids:
                assignee_ids.append(str(users[0].id))

        self._raw_issue.assignee_ids = assignee_ids
        self._raw_issue.save()

    def get_comment(self, comment_id: int) -> IssueComment:
        return GitlabIssueComment(self._raw_issue.notes.get(comment_id))

class GitlabIssueComment(GitlabComment, IssueComment):
    def __str__(self) -> str:
        return "Gitlab" + super().__str__()

class GitlabPRComment(GitlabComment, PRComment):
    def __str__(self) -> str:
        return "Gitlab" + super().__str__()

class GitlabProject(BaseGitProject):
    service: "ogr_gitlab.GitlabService"

    def __init__(
        self,
        repo: str,
        service: "ogr_gitlab.GitlabService",
        namespace: str,
        gitlab_repo=None,
        **unprocess_kwargs,
    ) -> None:
        if unprocess_kwargs:
            logger.warning(
                f"GitlabProject will not process these kwargs: {unprocess_kwargs}",
            )
        super().__init__(repo, service, namespace)
        self._gitlab_repo = gitlab_repo
        self.read_only = False

    @property
    def gitlab_repo(self) -> GitlabObjectsProject:
        if not self._gitlab_repo:
            self._gitlab_repo = self.service.gitlab_instance.projects.get(
                f"{self.namespace}/{self.repo}",
            )
        return self._gitlab_repo

    @property
    def is_fork(self) -> bool:
        return bool("forked_from_project" in self.gitlab_repo.attributes)

    @property
    def parent(self) -> Optional["GitlabProject"]:
        if self.is_fork:
            parent_dict = self.gitlab_repo.attributes["forked_from_project"]
            return GitlabProject(
                repo=parent_dict["path"],
                service=self.service,
                namespace=parent_dict["namespace"]["full_path"],
            )
        return None

    @property
    def default_branch(self) -> Optional[str]:
        return self.gitlab_repo.attributes.get("default_branch")

    def __str__(self) -> str:
        return f'GitlabProject(namespace="{self.namespace}", repo="{self.repo}")'

    def __eq__(self, o: object) -> bool:
        if not isinstance(o, GitlabProject):
            return False

        return (
            self.repo == o.repo
            and self.namespace == o.namespace
            and self.service == o.service
        )

    @property
    def has_issues(self) -> bool:
        return self.gitlab_repo.issues_enabled

    def _construct_fork_project(self) -> Optional["GitlabProject"]:
        user_login = self.service.user.get_username()
        try:
            project = GitlabProject(
                repo=self.repo,
                service=self.service,
                namespace=user_login,
            )
            if project.gitlab_repo:
                return project
        except Exception as ex:
            logger.debug(f"Project {user_login}/{self.repo} does not exist: {ex}")
        return None

    def exists(self) -> bool:
        try:
            _ = self.gitlab_repo
            return True
        except gitlab.exceptions.GitlabGetError as ex:
            if "404 Project Not Found" in str(ex):
                return False
            raise GitlabAPIException from ex

    def is_private(self) -> bool:
        return self.gitlab_repo.attributes["visibility"] == "private"

    def is_forked(self) -> bool:
        return bool(self._construct_fork_project())

    def get_description(self) -> str:
        return self.gitlab_repo.attributes["description"]

    @property
    def description(self) -> str:
        return self.gitlab_repo.attributes["description"]

    @description.setter
    def description(self, new_description: str) -> None:
        self.gitlab_repo.description = new_description
        self.gitlab_repo.save()

    def get_fork(self, create: bool = True) -> Optional["GitlabProject"]:
        username = self.service.user.get_username()
        for fork in self.get_forks():
            if fork.gitlab_repo.namespace["full_path"] == username:
                return fork

        if not self.is_forked():
            if create:
                return self.fork_create()

            logger.info(
                f"Fork of {self.gitlab_repo.attributes['name']}"
                " does not exist and we were asked not to create it.",
            )
            return None
        return self._construct_fork_project()

    def get_owners(self) -> list[str]:
        return self._get_collaborators_with_given_access(
            access_levels=[gitlab.const.OWNER_ACCESS],
        )

    def who_can_close_issue(self) -> set[str]:
        return set(
            self._get_collaborators_with_given_access(
                access_levels=[
                    gitlab.const.REPORTER_ACCESS,
                    gitlab.const.DEVELOPER_ACCESS,
                    gitlab.const.MAINTAINER_ACCESS,
                    gitlab.const.OWNER_ACCESS,
                ],
            ),
        )

    def who_can_merge_pr(self) -> set[str]:
        return set(
            self._get_collaborators_with_given_access(
                access_levels=[
                    gitlab.const.DEVELOPER_ACCESS,
                    gitlab.const.MAINTAINER_ACCESS,
                    gitlab.const.OWNER_ACCESS,
                ],
            ),
        )

    def can_merge_pr(self, username) -> bool:
        return username in self.who_can_merge_pr()

    def delete(self) -> None:
        self.gitlab_repo.delete()

    def _get_collaborators_with_given_access(
        self,
        access_levels: list[int],
    ) -> list[str]:
        """
        Get all project collaborators with one of the given access levels.
        Access levels:
            10 => Guest access
            20 => Reporter access
            30 => Developer access
            40 => Maintainer access
            50 => Owner access

        Returns:
            List of usernames.
        """
        # TODO: Remove once ‹members_all› is available for all releases of ogr
        all_members = None
        if hasattr(self.gitlab_repo, "members_all"):
            all_members = self.gitlab_repo.members_all.list(all=True)
        else:
            all_members = self.gitlab_repo.members.all(all=True)

        response = []
        for member in all_members:
            if isinstance(member, dict):
                access_level = member["access_level"]
                username = member["username"]
            else:
                access_level = member.access_level
                username = member.username
            if access_level in access_levels:
                response.append(username)
        return response

    def add_user(self, user: str, access_level: AccessLevel) -> None:
        access_dict = {
            AccessLevel.pull: gitlab.const.GUEST_ACCESS,
            AccessLevel.triage: gitlab.const.REPORTER_ACCESS,
            AccessLevel.push: gitlab.const.DEVELOPER_ACCESS,
            AccessLevel.admin: gitlab.const.MAINTAINER_ACCESS,
            AccessLevel.maintain: gitlab.const.OWNER_ACCESS,
        }
        try:
            user_id = self.service.gitlab_instance.users.list(username=user)[0].id
        except Exception as e:
            raise GitlabAPIException(f"User {user} not found") from e
        try:
            self.gitlab_repo.members.create(
                {"user_id": user_id, "access_level": access_dict[access_level]},
            )
        except Exception as e:
            raise GitlabAPIException(f"User {user} already exists") from e

    def request_access(self) -> None:
        try:
            self.gitlab_repo.accessrequests.create({})
        except gitlab.exceptions.GitlabCreateError as e:
            raise GitlabAPIException("Unable to request access") from e

    @indirect(GitlabPullRequest.get_list)
    def get_pr_list(self, status: PRStatus = PRStatus.open) -> list["PullRequest"]:
        pass

    def get_sha_from_tag(self, tag_name: str) -> str:
        try:
            tag = self.gitlab_repo.tags.get(tag_name)
            return tag.attributes["commit"]["id"]
        except gitlab.exceptions.GitlabGetError as ex:
            logger.error(f"Tag {tag_name} was not found.")
            raise GitlabAPIException(f"Tag {tag_name} was not found.") from ex

    @indirect(GitlabPullRequest.create)
    def create_pr(
        self,
        title: str,
        body: str,
        target_branch: str,
        source_branch: str,
        fork_username: Optional[str] = None,
    ) -> "PullRequest":
        pass

    def commit_comment(
        self,
        commit: str,
        body: str,
        filename: Optional[str] = None,
        row: Optional[int] = None,
    ) -> "CommitComment":
        try:
            commit_object: ProjectCommit = self.gitlab_repo.commits.get(commit)
        except gitlab.exceptions.GitlabGetError as ex:
            logger.error(f"Commit {commit} was not found.")
            raise GitlabAPIException(f"Commit {commit} was not found.") from ex

        if filename and row:
            raw_comment = commit_object.comments.create(
                {"note": body, "path": filename, "line": row, "line_type": "new"},
            )
        else:
            raw_comment = commit_object.comments.create({"note": body})
        return self._commit_comment_from_gitlab_object(raw_comment, commit)

    @staticmethod
    def _commit_comment_from_gitlab_object(raw_comment, commit) -> CommitComment:
        return CommitComment(
            sha=commit,
            body=raw_comment.note,
            author=raw_comment.author["username"],
        )

    def get_commit_comments(self, commit: str) -> list[CommitComment]:
        try:
            commit_object: ProjectCommit = self.gitlab_repo.commits.get(commit)
        except gitlab.exceptions.GitlabGetError as ex:
            logger.error(f"Commit {commit} was not found.")
            raise GitlabAPIException(f"Commit {commit} was not found.") from ex

        return [
            self._commit_comment_from_gitlab_object(comment, commit)
            for comment in commit_object.comments.list()
        ]

    @indirect(GitlabCommitFlag.set)
    def set_commit_status(
        self,
        commit: str,
        state: Union[CommitStatus, str],
        target_url: str,
        description: str,
        context: str,
        trim: bool = False,
    ) -> "CommitFlag":
        pass

    @indirect(GitlabCommitFlag.get)
    def get_commit_statuses(self, commit: str) -> list[CommitFlag]:
        pass

    def get_git_urls(self) -> dict[str, str]:
        return {
            "git": self.gitlab_repo.attributes["http_url_to_repo"],
            "ssh": self.gitlab_repo.attributes["ssh_url_to_repo"],
        }

    def fork_create(self, namespace: Optional[str] = None) -> "GitlabProject":
        data = {}
        if namespace:
            data["namespace_path"] = namespace

        try:
            fork = self.gitlab_repo.forks.create(data=data)
        except gitlab.GitlabCreateError as ex:
            logger.error(f"Repo {self.gitlab_repo} cannot be forked")
            raise GitlabAPIException(
                f"Repo {self.gitlab_repo} cannot be forked",
            ) from ex
        logger.debug(f"Forked to {fork.namespace['full_path']}/{fork.path}")
        return GitlabProject(
            namespace=fork.namespace["full_path"],
            service=self.service,
            repo=fork.path,
        )

    def change_token(self, new_token: str):
        self.service.change_token(new_token)

    def get_branches(self) -> list[str]:
        return [branch.name for branch in self.gitlab_repo.branches.list(all=True)]

    def get_file_content(self, path, ref=None) -> str:
        ref = ref or self.default_branch
        # GitLab cannot resolve './'
        path = os.path.normpath(path)
        try:
            file = self.gitlab_repo.files.get(file_path=path, ref=ref)
            return file.decode().decode()
        except gitlab.exceptions.GitlabGetError as ex:
            if ex.response_code == 404:
                raise FileNotFoundError(f"File '{path}' on {ref} not found") from ex
            raise GitlabAPIException() from ex

    def get_files(
        self,
        ref: Optional[str] = None,
        filter_regex: Optional[str] = None,
        recursive: bool = False,
    ) -> list[str]:
        ref = ref or self.default_branch
        paths = [
            file_dict["path"]
            for file_dict in self.gitlab_repo.repository_tree(
                ref=ref,
                recursive=recursive,
                all=True,
            )
            if file_dict["type"] != "tree"
        ]
        if filter_regex:
            paths = filter_paths(paths, filter_regex)

        return paths

    @indirect(GitlabIssue.get_list)
    def get_issue_list(
        self,
        status: IssueStatus = IssueStatus.open,
        author: Optional[str] = None,
        assignee: Optional[str] = None,
        labels: Optional[list[str]] = None,
    ) -> list[Issue]:
        pass

    @indirect(GitlabIssue.get)
    def get_issue(self, issue_id: int) -> Issue:
        pass

    @indirect(GitlabIssue.create)
    def create_issue(
        self,
        title: str,
        body: str,
        private: Optional[bool] = None,
        labels: Optional[list[str]] = None,
        assignees: Optional[list[str]] = None,
    ) -> Issue:
        pass

    @indirect(GitlabPullRequest.get)
    def get_pr(self, pr_id: int) -> PullRequest:
        pass

    def get_tags(self) -> list["GitTag"]:
        tags = self.gitlab_repo.tags.list()
        return [GitTag(tag.name, tag.commit["id"]) for tag in tags]

    def _git_tag_from_tag_name(self, tag_name: str) -> GitTag:
        git_tag = self.gitlab_repo.tags.get(tag_name)
        return GitTag(name=git_tag.name, commit_sha=git_tag.commit["id"])

    @indirect(GitlabRelease.get_list)
    def get_releases(self) -> list[Release]:
        pass

    @indirect(GitlabRelease.get)
    def get_release(self, identifier=None, name=None, tag_name=None) -> GitlabRelease:
        pass

    @indirect(GitlabRelease.create)
    def create_release(
        self,
        tag: str,
        name: str,
        message: str,
        commit_sha: Optional[str] = None,
    ) -> GitlabRelease:
        pass

    @indirect(GitlabRelease.get_latest)
    def get_latest_release(self) -> Optional[GitlabRelease]:
        pass

    def list_labels(self):
        """
        Get list of labels in the repository.

        Returns:
            List of labels in the repository.
        """
        return list(self.gitlab_repo.labels.list())

    def get_forks(self) -> list["GitlabProject"]:
        try:
            forks = self.gitlab_repo.forks.list()
        except KeyError as ex:
            # > item = self._data[self._current]
            # > KeyError: 0
            # looks like some API weirdness
            raise OperationNotSupported(
                "Please upgrade python-gitlab to a newer version.",
            ) from ex
        return [
            GitlabProject(
                repo=fork.path,
                namespace=fork.namespace["full_path"],
                service=self.service,
            )
            for fork in forks
        ]

    def update_labels(self, labels):
        """
        Update the labels of the repository. (No deletion, only add not existing ones.)

        Args:
            labels: List of labels to be added.

        Returns:
            Number of added labels.
        """
        current_label_names = [la.name for la in list(self.gitlab_repo.labels.list())]
        changes = 0
        for label in labels:
            if label.name not in current_label_names:
                color = self._normalize_label_color(color=label.color)
                self.gitlab_repo.labels.create(
                    {
                        "name": label.name,
                        "color": color,
                        "description": label.description or "",
                    },
                )

                changes += 1
        return changes

    @staticmethod
    def _normalize_label_color(color):
        if not color.startswith("#"):
            return f"#{color}"
        return color

    def get_web_url(self) -> str:
        return self.gitlab_repo.web_url

    def get_sha_from_branch(self, branch: str) -> Optional[str]:
        try:
            return self.gitlab_repo.branches.get(branch).attributes["commit"]["id"]
        except GitlabGetError as ex:
            if ex.response_code == 404:
                return None
            raise GitlabAPIException from ex

    def get_contributors(self) -> set[str]:
        """
        Returns:
            Unique authors of the commits in the project.
        """

        def format_contributor(contributor: dict[str, Any]) -> str:
            return f"{contributor['name']} <{contributor['email']}>"

        return set(
            map(format_contributor, self.gitlab_repo.repository_contributors(all=True)),
        )

    def users_with_write_access(self) -> set[str]:
        return set(
            self._get_collaborators_with_given_access(
                access_levels=[
                    gitlab.const.DEVELOPER_ACCESS,
                    gitlab.const.MAINTAINER_ACCESS,
                    gitlab.const.OWNER_ACCESS,
                ],
            ),
        )

class GitlabPullRequest(BasePullRequest):
    _raw_pr: _GitlabMergeRequest
    _target_project: "ogr_gitlab.GitlabProject"
    _source_project: "ogr_gitlab.GitlabProject" = None
    _merge_commit_status: ClassVar[dict[str, MergeCommitStatus]] = {
        "can_be_merged": MergeCommitStatus.can_be_merged,
        "cannot_be_merged": MergeCommitStatus.cannot_be_merged,
        "unchecked": MergeCommitStatus.unchecked,
        "checking": MergeCommitStatus.checking,
        "cannot_be_merged_recheck": MergeCommitStatus.cannot_be_merged_recheck,
    }

    @property
    def title(self) -> str:
        return self._raw_pr.title

    @title.setter
    def title(self, new_title: str) -> None:
        self._raw_pr.title = new_title
        self._raw_pr.save()

    @property
    def id(self) -> int:
        return self._raw_pr.iid

    @property
    def status(self) -> PRStatus:
        return (
            PRStatus.open
            if self._raw_pr.state == "opened"
            else PRStatus[self._raw_pr.state]
        )

    @property
    def url(self) -> str:
        return self._raw_pr.web_url

    @property
    def description(self) -> str:
        return self._raw_pr.description

    @description.setter
    def description(self, new_description: str) -> None:
        self._raw_pr.description = new_description
        self._raw_pr.save()

    @property
    def author(self) -> str:
        return self._raw_pr.author["username"]

    @property
    def source_branch(self) -> str:
        return self._raw_pr.source_branch

    @property
    def target_branch(self) -> str:
        return self._raw_pr.target_branch

    @property
    def created(self) -> datetime.datetime:
        return self._raw_pr.created_at

    @property
    def labels(self) -> list[PRLabel]:
        return [GitlabPRLabel(label, self) for label in self._raw_pr.labels]

    @property
    def diff_url(self) -> str:
        return f"{self._raw_pr.web_url}/diffs"

    @property
    def commits_url(self) -> str:
        return f"{self._raw_pr.web_url}/commits"

    @property
    def patch(self) -> bytes:
        response = requests.get(f"{self.url}.patch")

        if not response.ok:
            cls = OgrNetworkError if response.status_code >= 500 else GitlabAPIException
            raise cls(
                f"Couldn't get patch from {self.url}.patch because {response.reason}.",
            )

        return response.content

    @property
    def head_commit(self) -> str:
        return self._raw_pr.sha

    @property
    def merge_commit_sha(self) -> Optional[str]:
        # when merged => return merge_commit_sha
        # otherwise => return test merge if possible
        if self.status == PRStatus.merged:
            return self._raw_pr.merge_commit_sha

        # works for test merge only with python-gitlab>=2.10.0
        try:
            response = self._raw_pr.merge_ref()
        except GitlabGetError as ex:
            if ex.response_code == 400:
                return None
            raise
        return response.get("commit_id")

    @property
    def merge_commit_status(self) -> MergeCommitStatus:
        status = self._raw_pr.merge_status
        if status not in self._merge_commit_status:
            raise GitlabAPIException(f"Invalid merge_status {status}")

        return self._merge_commit_status[status]

    @property
    def source_project(self) -> "ogr_gitlab.GitlabProject":
        if self._source_project is None:
            self._source_project = (
                self._target_project.service.get_project_from_project_id(
                    self._raw_pr.attributes["source_project_id"],
                )
            )
        return self._source_project

    def __str__(self) -> str:
        return "Gitlab" + super().__str__()

    @staticmethod
    def create(
        project: "ogr_gitlab.GitlabProject",
        title: str,
        body: str,
        target_branch: str,
        source_branch: str,
        fork_username: Optional[str] = None,
    ) -> "PullRequest":
        """
        How to create PR:
        -  upstream -> upstream - call on upstream, fork_username unset
        -  fork -> upstream - call on fork, fork_username unset
           also can call on upstream with fork_username, not supported way of using
        -  fork -> fork - call on fork, fork_username set
        -  fork -> other_fork - call on fork, fork_username set to other_fork owner
        """
        repo = project.gitlab_repo
        parameters = {
            "source_branch": source_branch,
            "target_branch": target_branch,
            "title": title,
            "description": body,
        }
        target_id = None

        target_project = project
        if project.is_fork and fork_username is None:
            # handles fork -> upstream (called on fork)
            target_id = project.parent.gitlab_repo.attributes["id"]
            target_project = project.parent
        elif fork_username and fork_username != project.namespace:
            # handles fork -> upstream
            #   (username of fork owner specified by fork_username)
            # handles fork -> other_fork
            #   (username of other_fork owner specified by fork_username)

            other_project = GitlabPullRequest.__get_fork(
                fork_username,
                project if project.parent is None else project.parent,
            )

            target_id = other_project.gitlab_repo.attributes["id"]
            if project.parent is None:
                target_id = repo.attributes["id"]
                repo = other_project.gitlab_repo
        # otherwise handles PR from the same project to same project

        if target_id is not None:
            parameters["target_project_id"] = target_id

        mr = repo.mergerequests.create(parameters)
        return GitlabPullRequest(mr, target_project)

    @staticmethod
    def __get_fork(
        fork_username: str,
        project: "ogr_gitlab.GitlabProject",
    ) -> "ogr_gitlab.GitlabProject":
        """
        Returns forked project of a requested user. Internal method, in case the fork
        doesn't exist, raises GitlabAPIException.

        Args:
            fork_username: Username of a user that owns requested fork.
            project: Project to search forks of.

        Returns:
            Requested fork.

        Raises:
            GitlabAPIException, in case the fork doesn't exist.
        """
        forks = list(
            filter(
                lambda fork: fork.gitlab_repo.namespace["full_path"] == fork_username,
                project.get_forks(),
            ),
        )
        if not forks:
            raise GitlabAPIException("Requested fork doesn't exist")
        return forks[0]

    @staticmethod
    def get(project: "ogr_gitlab.GitlabProject", pr_id: int) -> "PullRequest":
        try:
            mr = project.gitlab_repo.mergerequests.get(pr_id)
        except gitlab.GitlabGetError as ex:
            raise GitlabAPIException(f"No PR with id {pr_id} found") from ex
        return GitlabPullRequest(mr, project)

    @staticmethod
    def get_list(
        project: "ogr_gitlab.GitlabProject",
        status: PRStatus = PRStatus.open,
    ) -> list["PullRequest"]:
        # Gitlab API has status 'opened', not 'open'
        # f"Calling a `list()` method without specifying `get_all=True` or "
        # f"`iterator=True` will return a maximum of 20 items. "
        mrs = project.gitlab_repo.mergerequests.list(
            state=status.name if status != PRStatus.open else "opened",
            order_by="updated_at",
            sort="desc",
            # gitlab 3.3 syntax was all=True
            get_all=True,
        )
        return [GitlabPullRequest(mr, project) for mr in mrs]

    def update_info(
        self,
        title: Optional[str] = None,
        description: Optional[str] = None,
    ) -> "PullRequest":
        if title:
            self._raw_pr.title = title
        if description:
            self._raw_pr.description = description

        self._raw_pr.save()
        return self

    def _get_all_comments(self) -> list[PRComment]:
        return [
            GitlabPRComment(parent=self, raw_comment=raw_comment)
            for raw_comment in self._raw_pr.notes.list(sort="asc", all=True)
        ]

    def get_all_commits(self) -> list[str]:
        return [commit.id for commit in self._raw_pr.commits()]

    def comment(
        self,
        body: str,
        commit: Optional[str] = None,
        filename: Optional[str] = None,
        row: Optional[int] = None,
    ) -> "PRComment":
        comment = self._raw_pr.notes.create({"body": body})
        return GitlabPRComment(parent=self, raw_comment=comment)

    def close(self) -> "PullRequest":
        self._raw_pr.state_event = "close"
        self._raw_pr.save()
        return self

    def merge(self) -> "PullRequest":
        self._raw_pr.merge()
        return self

    def add_label(self, *labels: str) -> None:
        self._raw_pr.labels += labels
        self._raw_pr.save()

    def get_comment(self, comment_id: int) -> PRComment:
        return GitlabPRComment(self._raw_pr.notes.get(comment_id))

class GitlabRelease(Release):
    _raw_release: _GitlabRelease
    project: "ogr_gitlab.GitlabProject"

    @property
    def title(self):
        return self._raw_release.name

    @property
    def body(self):
        return self._raw_release.description

    @property
    def git_tag(self) -> GitTag:
        return self.project._git_tag_from_tag_name(self.tag_name)

    @property
    def tag_name(self) -> str:
        return self._raw_release.tag_name

    @property
    def url(self) -> Optional[str]:
        return f"{self.project.get_web_url()}/-/releases/{self.tag_name}"

    @property
    def created_at(self) -> datetime.datetime:
        return self._raw_release.created_at

    @property
    def tarball_url(self) -> str:
        return self._raw_release.assets["sources"][1]["url"]

    def __str__(self) -> str:
        return "Gitlab" + super().__str__()

    @staticmethod
    def get(
        project: "ogr_gitlab.GitlabProject",
        identifier: Optional[int] = None,
        name: Optional[str] = None,
        tag_name: Optional[str] = None,
    ) -> "Release":
        release = project.gitlab_repo.releases.get(tag_name)
        return GitlabRelease(release, project)

    @staticmethod
    def get_latest(project: "ogr_gitlab.GitlabProject") -> Optional["Release"]:
        releases = project.gitlab_repo.releases.list()
        # list of releases sorted by released_at
        return GitlabRelease(releases[0], project) if releases else None

    @staticmethod
    def get_list(project: "ogr_gitlab.GitlabProject") -> list["Release"]:
        if not hasattr(project.gitlab_repo, "releases"):
            raise OperationNotSupported(
                "This version of python-gitlab does not support release, please upgrade.",
            )
        releases = project.gitlab_repo.releases.list(all=True)
        return [GitlabRelease(release, project) for release in releases]

    @staticmethod
    def create(
        project: "ogr_gitlab.GitlabProject",
        tag: str,
        name: str,
        message: str,
        ref: Optional[str] = None,
    ) -> "Release":
        release = project.gitlab_repo.releases.create(
            {"name": name, "tag_name": tag, "description": message, "ref": ref},
        )
        return GitlabRelease(release, project)

    def edit_release(self, name: str, message: str) -> None:
        raise OperationNotSupported("edit_release not supported on GitLab")

@use_for_service("gitlab")  # anything containing a gitlab word in hostname
# + list of community-hosted instances based on the following list
# https://wiki.p2pfoundation.net/List_of_Community-Hosted_GitLab_Instances
@use_for_service("salsa.debian.org")
@use_for_service("git.fosscommunity.in")
@use_for_service("framagit.org")
@use_for_service("dev.gajim.org")
@use_for_service("git.coop")
@use_for_service("lab.libreho.st")
@use_for_service("git.linux-kernel.at")
@use_for_service("git.pleroma.social")
@use_for_service("git.silence.dev")
@use_for_service("code.videolan.org")
@use_for_service("source.puri.sm")
class GitlabService(BaseGitService):
    name = "gitlab"

    def __init__(self, token=None, instance_url=None, ssl_verify=True, **kwargs):
        super().__init__(token=token)
        self.instance_url = instance_url or "https://gitlab.com"
        self.token = token
        self.ssl_verify = ssl_verify
        self._gitlab_instance = None

        if kwargs:
            logger.warning(f"Ignored keyword arguments: {kwargs}")

    @property
    def gitlab_instance(self) -> gitlab.Gitlab:
        if not self._gitlab_instance:
            self._gitlab_instance = gitlab.Gitlab(
                url=self.instance_url,
                private_token=self.token,
                ssl_verify=self.ssl_verify,
            )
            if self.token:
                self._gitlab_instance.auth()
        return self._gitlab_instance

    @property
    def user(self) -> GitUser:
        return GitlabUser(service=self)

    def __str__(self) -> str:
        token_str = (
            f", token='{self.token[:1]}***{self.token[-1:]}'" if self.token else ""
        )
        ssl_str = ", ssl_verify=False" if not self.ssl_verify else ""
        return (
            f"GitlabService(instance_url='{self.instance_url}'"
            f"{token_str}"
            f"{ssl_str})"
        )

    def __eq__(self, o: object) -> bool:
        if not issubclass(o.__class__, GitlabService):
            return False

        return (
            self.token == o.token  # type: ignore
            and self.instance_url == o.instance_url  # type: ignore
            and self.ssl_verify == o.ssl_verify  # type: ignore
        )

    def __hash__(self) -> int:
        return hash(str(self))

    def get_project(
        self,
        repo=None,
        namespace=None,
        is_fork=False,
        **kwargs,
    ) -> "GitlabProject":
        if is_fork:
            namespace = self.user.get_username()
        return GitlabProject(repo=repo, namespace=namespace, service=self, **kwargs)

    def get_project_from_project_id(self, iid: int) -> "GitlabProject":
        gitlab_repo = self.gitlab_instance.projects.get(iid)
        return GitlabProject(
            repo=gitlab_repo.attributes["path"],
            namespace=gitlab_repo.attributes["namespace"]["full_path"],
            service=self,
            gitlab_repo=gitlab_repo,
        )

    def change_token(self, new_token: str) -> None:
        self.token = new_token
        self._gitlab_instance = None

    def project_create(
        self,
        repo: str,
        namespace: Optional[str] = None,
        description: Optional[str] = None,
    ) -> "GitlabProject":
        data = {"name": repo}
        if namespace:
            try:
                group = self.gitlab_instance.groups.get(namespace)
            except gitlab.GitlabGetError as ex:
                raise GitlabAPIException(f"Group {namespace} not found.") from ex
            data["namespace_id"] = group.id

        if description:
            data["description"] = description
        try:
            new_project = self.gitlab_instance.projects.create(data)
        except gitlab.GitlabCreateError as ex:
            raise GitlabAPIException("Project already exists") from ex
        return GitlabProject(
            repo=repo,
            namespace=namespace,
            service=self,
            gitlab_repo=new_project,
        )

    def list_projects(
        self,
        namespace: Optional[str] = None,
        user: Optional[str] = None,
        search_pattern: Optional[str] = None,
        language: Optional[str] = None,
    ) -> list[GitProject]:
        if namespace:
            group = self.gitlab_instance.groups.get(namespace)
            projects = group.projects.list(all=True)
        elif user:
            user_object = self.gitlab_instance.users.list(username=user)[0]
            projects = user_object.projects.list(all=True)
        else:
            raise OperationNotSupported

        if language:
            # group.projects.list gives us a GroupProject instance
            # in order to be able to filter by language we need Project instance
            projects_to_convert = [
                self.gitlab_instance.projects.get(item.attributes["id"])
                for item in projects
                if language
                in self.gitlab_instance.projects.get(item.attributes["id"]).languages()
            ]
        else:
            projects_to_convert = projects
        return [
            GitlabProject(
                repo=project.attributes["path"],
                namespace=project.attributes["namespace"]["full_path"],
                service=self,
            )
            for project in projects_to_convert
        ]

class GitlabUser(BaseGitUser):
    service: "ogr_gitlab.GitlabService"

    def __init__(self, service: "ogr_gitlab.GitlabService") -> None:
        super().__init__(service=service)

    def __str__(self) -> str:
        return f'GitlabUser(username="{self.get_username()}")'

    @property
    def _gitlab_user(self):
        return self.service.gitlab_instance.user

    def get_username(self) -> str:
        return self._gitlab_user.username

    def get_email(self) -> str:
        return self._gitlab_user.email

    def get_projects(self) -> list["ogr_gitlab.GitlabProject"]:
        raise OperationNotSupported

    def get_forks(self) -> list["ogr_gitlab.GitlabProject"]:
        raise OperationNotSupported