はじめに
現場がJiraを使っていて、気に入ったからAPIの使い方をさらっておきたい。
2025では 公式ドキュメント を見て REST-API で書いています。ライブラリを入れなくてもよくなったみたい。
TODO: CRUDの作成
参考
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