はじめに

現場がJiraを使っていて、気に入ったからAPIの使い方をさらっておきたい。

2025では　公式ドキュメント　を見て REST-API で書いています。ライブラリを入れなくてもよくなったみたい。

TODO: CRUDの作成

参考

JIRA REST API reference v3

Jira側でAPIトークンの生成

source

.env.example

  :
EMAIL_HOST_USER=abc@example.com
JIRA_YOUR_DOMAIN=
JIRA_API_KEY=
  :

lib/jira/jira_service.py

import os

import requests
from requests import HTTPError
from requests.auth import HTTPBasicAuth

from lib.jira.valueobject.ticket import (
    ProjectVO,
    IssueVO,
    SubTaskVO,
    # CreateIssuePayload,
)


class JiraService:
    """
    Service class for interacting with the JIRA API.
    """

    def __init__(self, domain: str = None, email: str = None, api_token: str = None):
        """
        Initialize the JiraService with domain, email, and API token.

        Args:
            domain (str): The JIRA domain (e.g., "your-domain").
            email (str): The email address for authentication.
            api_token (str): The API token for authentication.
        """
        missing_envs = []
        if not domain:
            missing_envs.append("YOUR_DOMAIN")
        if not email:
            missing_envs.append("EMAIL")
        if not api_token:
            missing_envs.append("API_TOKEN")

        if missing_envs:
            error_message = f"400 Bad Request: Missing required environment variable(s): {', '.join(missing_envs)}"
            raise HTTPError(error_message)

        self.base_url = f"https://{domain}.atlassian.net"
        # TODO: Basic認証の問題なのか401は出ない。OAuth 2.0化が必要か
        self.auth = HTTPBasicAuth(email, api_token)
        self.headers = {"Accept": "application/json"}

    def fetch_projects(self) -> list[ProjectVO]:
        """
        Fetch all projects from the JIRA API using `isLast` for termination.

        Returns:
            List[ProjectVO]: A list of ProjectVOs containing project key and name.

        Raises:
            HTTPError: If the HTTP request returns an error response.
        """
        url = f"{self.base_url}/rest/api/3/project/search"
        all_projects = []

        while url:
            response = requests.get(url, headers=self.headers, auth=self.auth)

            # Automatically raise an exception for HTTP error responses
            response.raise_for_status()

            # Parse the response JSON
            data = response.json()

            # Extract the necessary fields from each project in the 'values'
            for project in data.get("values", []):
                all_projects.append(
                    ProjectVO(
                        key=project.get("key", "invalid key"),
                        name=project.get("name", "invalid name"),
                    )
                )

            # Exit the loop if this is the last page
            if data.get("isLast", False):
                break

            # Update the URL for the next page
            url = data.get("nextPage", None)

        return all_projects

    def fetch_issues(self, project_key: str) -> dict[str, list[IssueVO]]:
        """
        Fetch all issues for a given project from the JIRA API.

        Args:
            project_key (str): The key of the project (e.g., "HSP").

        Returns:
            Dict[str, List[IssueVO]]: A dictionary with project keys as keys and lists of issues as values.

        Raises:
            HTTPError: If the HTTP request returns an error response.
        """
        url = f"{self.base_url}/rest/api/3/search"
        query = {
            "jql": f"project = {project_key} AND issuetype = Task",
            "maxResults": 50,  # Adjust as needed
            "fields": "key,issuetype,priority,assignee,status,description,summary,subtasks",
        }
        issues_by_project = {}

        while url:
            response = requests.get(
                url, headers=self.headers, auth=self.auth, params=query
            )

            # Automatically raise an exception for HTTP error responses
            response.raise_for_status()

            # Parse the response JSON
            data = response.json()

            # Extract issues
            for issue in data.get("issues", []):
                fields = issue.get("fields", {})
                issue_key = issue.get("key", "No issue key")

                # Check if the current issue is a regular issue or a subtask
                is_subtask = fields.get("issuetype", {}).get("subtask", False)
                issue_name = fields.get("summary", "No summary")
                issue_status = fields.get("status", {}).get("name", "No status")
                issue_priority = fields.get("priority", {}).get("name", "No priority")

                # "assignee" の処理: None なら None, そうでなければ displayName 取得
                issue_assignee = None
                if fields.get("assignee"):
                    issue_assignee = fields.get("assignee").get(
                        "displayName", "Unassigned"
                    )

                issue_description = "No description"

                # `description` が辞書形式の場合、正しく _parse_content_field を呼ぶ
                raw_description = fields.get("description")
                if isinstance(raw_description, dict) and raw_description.get("content"):
                    issue_description = self._parse_content_field(
                        raw_description.get("content")
                    )

                # If it's not a subtask, process as a regular issue (IssueVO)
                if not is_subtask:
                    sub_tasks = []
                    for sub_task in fields.get("subtasks", []):
                        sub_task_key = sub_task.get("key", "No sub-task")
                        sub_task_name = sub_task.get("fields").get(
                            "summary", "No summary"
                        )
                        sub_task_status = (
                            sub_task.get("fields", {})
                            .get("status", {})
                            .get("name", "No status")
                        )
                        sub_task_priority = (
                            sub_task.get("fields", {})
                            .get("priority", {})
                            .get("name", "No priority")
                        )
                        sub_tasks.append(
                            SubTaskVO(
                                key=sub_task_key,
                                name=sub_task_name,
                                status=sub_task_status,
                                priority=sub_task_priority,
                            )
                        )

                    issue_obj = IssueVO(
                        key=issue_key,
                        name=issue_name,
                        description=issue_description,
                        priority=issue_priority,
                        assignee=issue_assignee,
                        status=issue_status,
                        sub_tasks=sub_tasks,
                    )

                    # Add the issue object to the dictionary under the project key
                    if project_key not in issues_by_project:
                        issues_by_project[project_key] = []
                    issues_by_project.get(project_key).append(issue_obj)

            # Update the URL for the next page if available
            if "nextPage" in data:
                url = data["nextPage"]
            else:
                url = None

        return issues_by_project

    @staticmethod
    def _parse_content_field(content: list) -> str:
        """
        Helper function to extract text from the 'content' field in the description.

        Args:
            content (list): The 'content' field from the issue description.

        Returns:
            str: Extracted plain text description.
        """
        description_text = []

        for block in content:
            # Ensure the block has "type" and "content"
            if not isinstance(block, dict) or "content" not in block:
                continue

            for inner_block in block.get("content", []):
                # Look for 'text' in inner blocks
                if isinstance(inner_block, dict) and inner_block.get("type") == "text":
                    description_text.append(inner_block.get("text", ""))

        return " ".join(description_text)

    # def create_issue(self, payload: CreateIssuePayload):
    #     """
    #     JIRAチケットを作成する
    #
    #     Args:
    #         payload (CreateIssuePayload): ペイロードデータを表すValue Object
    #
    #     Returns:
    #         dict: 作成されたチケットの詳細
    #     """
    #     url = f"{self.base_url}/rest/api/3/issue"
    #
    #     # POSTリクエストを送信
    #     response = requests.post(
    #         url,
    #         data=json.dumps(payload.to_dict()),
    #         headers=self.headers,
    #         auth=self.auth,
    #     )
    #
    #     if response.status_code != 201:  # 201は「作成成功」を表す
    #         raise HTTPError(
    #             f"Failed to create issue: {response.status_code} {response.text}"
    #         )
    #
    #     return response.json()  # 作成されたチケットのJSONレスポンスを返す


if __name__ == "__main__":
    # API docs: https://developer.atlassian.com/cloud/jira/platform/rest/v3/intro/#about

    # JIRA configuration: Replace with actual values or environment variables
    YOUR_DOMAIN = os.getenv("JIRA_YOUR_DOMAIN")
    EMAIL = os.getenv("EMAIL_HOST_USER")
    API_TOKEN = os.getenv("JIRA_API_KEY")

    jira_service = JiraService(domain=YOUR_DOMAIN, email=EMAIL, api_token=API_TOKEN)

    try:
        for pjt in jira_service.fetch_projects():
            print(pjt)
            issues = jira_service.fetch_issues(project_key=pjt.key)

            for project_xxx, project_issues in issues.items():
                print(f"Project: {project_xxx}")
                for issue_xxx in project_issues:
                    print(issue_xxx)

        print("process done")
    except requests.exceptions.HTTPError as http_err:
        print(f"[HTTP Error] {http_err}")

    # # TODO: チケットを作成する機能を作る　https://developer.atlassian.com/cloud/jira/platform/rest/v3/api-group-issues/#api-rest-api-3-issue-post
    # payload_xxx = CreateIssuePayload(
    #     description_text="Order entry fails when selecting supplier.",
    #     issue_type_id="10000",
    #     labels=["bugfix"],
    #     parent_key="HEN",
    #     priority_id="20000",
    #     project_id="10000",
    #     reporter_id="5b10a2844c20165700ede21g",
    #     summary="Main order flow broken",
    # )
    # # チケットを作成
    # try:
    #     result = jira_service.create_issue(payload_xxx)
    #     print("チケットが作成されました:", result)
    # except HTTPError as e:
    #     print("エラーが発生しました:", str(e))
    #
    # # TODO: チケットを削除する
    # # TODO: チケットを編集する

lib/jira/test_jira_service.py

import unittest
from unittest.mock import patch, Mock

from requests import HTTPError

from lib.jira.jira_service import JiraService
from lib.jira.valueobject.ticket import IssueVO


class TestJiraServiceFetchIssues(unittest.TestCase):
    """
    JiraServiceクラスのfetch_issuesメソッドに関する単体テストを行います。
    - 各テストケースでは、APIのモックレスポンスを利用して、fetch_issuesの期待される動作を検証します。
    """

    def setUp(self):
        """
        各テストの実行前に共通のセットアップ処理を行います。
        JiraServiceのインスタンスを生成します。
        """
        self.service = JiraService(
            domain="test-domain", email="test@example.com", api_token="test-token"
        )

    @staticmethod
    def create_mock_response(json_data, status_code=200):
        """
        APIレスポンスをモックとして作成する共通のヘルパーメソッド。

        Args:
            json_data (dict): モックレスポンスとして返すJSONデータ
            status_code (int): HTTPステータスコード（デフォルトは200）

        Returns:
            Mock: モック化したレスポンスオブジェクト
        """
        mock_response = Mock()
        mock_response.json.return_value = json_data
        mock_response.status_code = status_code
        mock_response.raise_for_status = (
            Mock()
        )  # ステータスコードに応じた例外を発生させるモックを設定
        return mock_response

    @patch("requests.get")
    def test_fetch_issues_with_assignee(self, mock_get):
        """
        テスト内容:
            fetch_issuesが、「assignee」が指定されている課題を処理できることを確認する。
        シナリオ:
            - APIレスポンスに "assignee" フィールドが存在し、"displayName" を持つ場合をテストする。
            - "assignee" の情報が正しくIssueVOに設定されていることを検証する。
        """
        mock_get.return_value = self.create_mock_response(
            {
                "issues": [
                    {
                        "key": "HEN-1",
                        "fields": {
                            "summary": "Test Issue 1",
                            "description": None,
                            "assignee": {"displayName": "John Doe"},
                            "priority": {"name": "High"},
                            "status": {"name": "To Do"},
                            "subtasks": [],
                        },
                    }
                ]
            }
        )

        # 実行
        issues = self.service.fetch_issues("HEN")

        # 検証
        self.assertEqual(len(issues["HEN"]), 1)
        issue = issues["HEN"][0]
        self.assertIsInstance(issue, IssueVO)
        self.assertEqual(issue.assignee, "John Doe")
        self.assertEqual(issue.name, "Test Issue 1")
        self.assertEqual(issue.description, "No description")  # デフォルトの説明
        self.assertEqual(issue.priority, "High")
        self.assertEqual(issue.status, "To Do")

    @patch("requests.get")
    def test_fetch_issues_without_assignee(self, mock_get):
        """
        テスト内容:
            fetch_issuesが、「assignee」が指定されていない（None）の課題を処理できることを確認する。
        シナリオ:
            - APIレスポンスに "assignee" フィールドがNoneの場合をテストする。
            - IssueVOの「assignee」の値がNoneとなっていることを検証する。
        """
        mock_get.return_value = self.create_mock_response(
            {
                "issues": [
                    {
                        "key": "HEN-2",
                        "fields": {
                            "summary": "Test Issue 2",
                            "description": None,
                            "assignee": None,
                            "priority": {"name": "Medium"},
                            "status": {"name": "In Progress"},
                            "subtasks": [],
                        },
                    }
                ]
            }
        )

        # 実行
        issues = self.service.fetch_issues("HEN")

        # 検証
        self.assertEqual(len(issues["HEN"]), 1)
        issue = issues["HEN"][0]
        self.assertIsInstance(issue, IssueVO)
        self.assertIsNone(issue.assignee)  # AssigneeがNoneであることを検証
        self.assertEqual(issue.name, "Test Issue 2")
        self.assertEqual(issue.description, "No description")  # デフォルトの説明
        self.assertEqual(issue.priority, "Medium")
        self.assertEqual(issue.status, "In Progress")

    @patch("requests.get")
    def test_fetch_issues_with_multiple_issues(self, mock_get):
        """
        テスト内容:
            fetch_issuesが、複数の課題を正しく処理できることを確認する。
        シナリオ:
            - APIレスポンスに複数の課題が含まれる場合をテストする。
            - 各課題のフィールド（assignee, priority, statusなど）が正しく設定されていることを検証する。
        """
        mock_get.return_value = self.create_mock_response(
            {
                "issues": [
                    {
                        "key": "HEN-3",
                        "fields": {
                            "summary": "Issue 1",
                            "assignee": {"displayName": "Jane Smith"},
                            "priority": {"name": "Low"},
                            "status": {"name": "In Progress"},
                            "subtasks": [],
                        },
                    },
                    {
                        "key": "HEN-4",
                        "fields": {
                            "summary": "Issue 2",
                            "assignee": None,
                            "priority": {"name": "High"},
                            "status": {"name": "To Do"},
                            "subtasks": [],
                        },
                    },
                ]
            }
        )

        # 実行
        issues = self.service.fetch_issues("HEN")

        # 検証
        self.assertEqual(len(issues["HEN"]), 2)

        # 課題1
        issue1 = issues["HEN"][0]
        self.assertEqual(issue1.assignee, "Jane Smith")
        self.assertEqual(issue1.name, "Issue 1")
        self.assertEqual(issue1.priority, "Low")
        self.assertEqual(issue1.status, "In Progress")

        # 課題2
        issue2 = issues["HEN"][1]
        self.assertIsNone(issue2.assignee)  # AssigneeがNoneであることを検証
        self.assertEqual(issue2.name, "Issue 2")
        self.assertEqual(issue2.priority, "High")
        self.assertEqual(issue2.status, "To Do")

    @patch("requests.get")
    def test_fetch_issues_with_invalid_project_key(self, mock_get):
        """
        fetch_issuesに無効なプロジェクトキーを渡した場合に正しく例外をスローするかを確認する。
        """
        # モックAPIの404エラー設定
        mock_response = self.create_mock_response({}, status_code=404)
        mock_response.raise_for_status.side_effect = HTTPError(
            "404 Client Error: Not Found"
        )
        mock_get.return_value = mock_response

        # 実行と例外の検証
        with self.assertRaises(HTTPError) as context:
            self.service.fetch_issues("INVALID")  # 存在しないプロジェクトキー

        # エラーメッセージを確認
        self.assertIn("404 Client Error", str(context.exception))
        mock_get.assert_called_once()

    def test_missing_base_url(self):
        """
        base_urlが空の場合にHTTPErrorがスローされ、エラーメッセージに正しい環境変数名が表示されることを確認する。
        """
        with self.assertRaises(HTTPError) as context:
            JiraService(domain=None, email="test@example.com", api_token="test_token")
        self.assertIn("YOUR_DOMAIN", str(context.exception))

    def test_missing_email(self):
        """
        emailが空の場合にHTTPErrorがスローされ、エラーメッセージに正しい環境変数名が表示されることを確認する。
        """
        with self.assertRaises(HTTPError) as context:
            JiraService(domain="test", email=None, api_token="test_token")
        self.assertIn("EMAIL", str(context.exception))

    def test_missing_api_token(self):
        """
        api_tokenが空の場合にHTTPErrorがスローされ、エラーメッセージに正しい環境変数名が表示されることを確認する。
        """
        with self.assertRaises(HTTPError) as context:
            JiraService(
                domain="test",
                email="test@example.com",
                api_token=None,
            )
        self.assertIn("API_TOKEN", str(context.exception))

    def test_multiple_missing_env_vars(self):
        """
        複数の環境変数が空の場合に、HTTPErrorがスローされ、不足している環境変数名がすべて表示されることを確認する。
        """
        with self.assertRaises(HTTPError) as context:
            JiraService(domain=None, email=None, api_token="test_token")
        error_message = str(context.exception)
        self.assertIn("YOUR_DOMAIN", error_message)
        self.assertIn("EMAIL", error_message)

    def test_all_env_vars_present(self):
        """
        すべての引数が正しく設定されている場合、サービスが正常に初期化されることを確認する。
        """
        try:
            service = JiraService(
                domain="test",
                email="test@example.com",
                api_token="test_token",
            )
            self.assertIsNotNone(service)
        except HTTPError:
            self.fail(
                "HTTPErrorが発生しました。すべての引数が存在している場合でも失敗しています。"
            )

lib/jira/valueobject/ticket.py

from dataclasses import dataclass


@dataclass
class ProjectVO:
    """
    Value Object (VO) to represent a JIRA project.

    Attributes:
        key (str): The unique key of the project.
        name (str): The name of the project.
    """

    key: str
    name: str


@dataclass
class SubTaskVO:
    """
    Data class to represent sub-tasks of an issue.

    Attributes:
        key (str): The unique key of the sub-task.
        name (str): The name of the sub-task.
        status (str): The current status of the sub-task.
        priority (str): The priority of the sub-task.
    """

    key: str
    name: str
    status: str
    priority: str


@dataclass
class IssueVO:
    """
    Data class to represent an issue.

    Attributes:
        key (str): The unique key of the issue.
        name (str): The title or summary of the issue.
        description (str): The detailed description of the issue.
        priority (str): The priority of the issue.
        assignee (str): The display name of the assigned user.
        status (str): The current status of the issue.
        sub_tasks (list[SubTaskVO]): The list of sub-tasks associated with the issue.
    """

    key: str
    name: str
    description: str
    priority: str
    assignee: str
    status: str
    sub_tasks: list[SubTaskVO]


class CreateIssuePayload:
    def __init__(
        self,
        description_text: str,
        issue_type_id: str,
        labels: list,
        parent_key: str,
        priority_id: str,
        project_id: str,
        reporter_id: str,
        summary: str,
    ):
        """
        チケット作成ペイロードデータのValue Object

        Args:
            description_text (str): チケットの説明文
            issue_type_id (str): イシュ―タイプのID
            labels (list): ラベル一覧
            parent_key (str): 親チケットのキー
            priority_id (str): 優先度のID
            project_id (str): プロジェクトのID
            reporter_id (str): 報告者のID
            summary (str): チケットの概要
        """
        self.description_text = description_text
        self.issue_type_id = issue_type_id
        self.labels = labels
        self.parent_key = parent_key
        self.priority_id = priority_id
        self.project_id = project_id
        self.reporter_id = reporter_id
        self.summary = summary

    def to_dict(self) -> dict:
        """
        ペイロードをJira API形式の辞書データに変換

        Returns:
            dict: Jira用のペイロードデータ
        """
        return {
            "fields": {
                "description": self._format_description(),
                "issuetype": {"id": self.issue_type_id},
                "labels": self.labels,
                "parent": {"key": self.parent_key},
                "priority": {"id": self.priority_id},
                "project": {"id": self.project_id},
                "reporter": {"id": self.reporter_id},
                "summary": self.summary,
            }
        }

    def _format_description(self) -> dict:
        """
        descriptionフィールドをJira標準のリッチテキスト形式に変換

        Returns:
            dict: リッチテキスト形式のdescriptionフィールド
        """
        return {
            "type": "doc",
            "version": 1,
            "content": [
                {
                    "type": "paragraph",
                    "content": [
                        {
                            "type": "text",
                            "text": self.description_text,
                        }
                    ],
                }
            ],
        }

確認

3つチケットがある

console

ProjectVO(key='HEN', name='henojiya')
Project: HEN
IssueVO(key='HEN-18', name='テストチケット', description='テストチケットの説明', priority='Medium', assignee='yoshitaka okada', status='To Do', sub_tasks=[SubTaskVO(key='HEN-19', name='サブタスク1', status='To Do', priority='Medium')])
IssueVO(key='HEN-4', name='まだ先の作業', description='No description', priority='Medium', assignee=None, status='To Do', sub_tasks=[])
IssueVO(key='HEN-3', name='レビュー', description='No description', priority='Medium', assignee='yoshitaka okada', status='完了', sub_tasks=[])
process done