Add pj

Author: mdellweg

.gitignore

@@ -1 +1,2 @@
 __pycache__/
+.mypy_cache/

pj/.gitignore

@@ -0,0 +1 @@
+.jiraauth

pj/pj.py

@@ -0,0 +1,244 @@
+#!/bin/env python3
+
+# Reference: https://jira.readthedocs.io
+
+import typing as t
+from pathlib import Path
+import json
+import tomllib
+
+from jira import JIRA
+from jira.utils import remove_empty_attributes
+from pydantic.dataclasses import dataclass
+import click
+
+
+@dataclass
+class Config:
+    server: str = "https://issues.redhat.com"
+    token: str = ""
+    project: str = "PULP"
+
+
+def read_config() -> Config:
+    conf_path = Path(".") / ".jiraauth"
+    data = tomllib.loads(conf_path.read_text())["default"]
+    return Config(**data)
+
+
+def search_issues_paginated(jira: JIRA, jql: str):
+    start_at = 0
+    max_results = 50
+    while results := jira.search_issues(
+        jql,
+        maxResults=max_results,
+        startAt=start_at,
+    ):
+        yield from results
+        start_at += max_results
+
+
+@click.group()
+@click.pass_context
+def main(ctx: click.Context) -> None:
+    config = read_config()
+    jira = JIRA(server=config.server, token_auth=config.token)
+    ctx.obj = {
+        "jira": jira,
+        "project": jira.project(config.project),
+    }
+
+
+@main.command()
+@click.pass_context
+def issues(ctx: click.Context) -> None:
+    jira = ctx.obj["jira"]
+    project = ctx.obj["project"]
+    for issue in search_issues_paginated(
+        jira,
+        f"project = {project} AND resolution = Unresolved ORDER BY priority DESC, updated DESC",
+    ):
+        print(issue, issue.fields.summary)
+
+
+@main.command()
+@click.pass_context
+def blocker(ctx: click.Context) -> None:
+    jira = ctx.obj["jira"]
+    project = ctx.obj["project"]
+    for issue in jira.search_issues(
+        f"project = {project} AND resolution = Unresolved AND priority = blocker ORDER BY updated DESC"
+    ):
+        print(issue, issue.fields.summary)
+
+
+@main.command()
+@click.pass_context
+def my_issues(ctx: click.Context) -> None:
+    jira = ctx.obj["jira"]
+    for issue in jira.search_issues(
+        "assignee = currentUser() AND resolution = Unresolved order by updated DESC"
+    ):
+        print(issue, issue.fields.summary)
+
+
+@main.command()
+def my_next_issue() -> None:
+    print("Who do you think I am?")
+
+
+@main.command()
+@click.argument("search_phrase")
+@click.pass_context
+def search(
+    ctx: click.Context,
+    search_phrase: str,
+) -> None:
+    jira = ctx.obj["jira"]
+    project = ctx.obj["project"]
+    jql = f"project = {project} AND resolution = Unresolved AND text ~ '{search_phrase}' ORDER BY updated DESC"
+    for issue in jira.search_issues(jql):
+        print(issue, issue.fields.summary)
+
+
+@main.command()
+@click.option("--raw", is_flag=True, default=False)
+@click.argument("issue_id")
+@click.pass_context
+def show(
+    ctx: click.Context,
+    raw: bool,
+    issue_id: str,
+) -> None:
+    jira = ctx.obj["jira"]
+    issue = jira.issue(issue_id)
+    if raw:
+        raw_issue = issue.raw
+        raw_issue["fields"] = remove_empty_attributes(raw_issue["fields"])
+        raw_issue = remove_empty_attributes(raw_issue)
+        print(json.dumps(raw_issue))
+    else:
+        print(issue)
+        print(issue.fields.summary)
+        print(issue.fields.description)
+        print("Status:", issue.fields.status.name)
+        fields = jira.fields()
+        for fieldname in ["Story Points", "Resolution"]:
+            field_id = next(
+                (field["id"] for field in fields if field["name"] == fieldname)
+            )
+            print(fieldname + ":", issue.get_field(field_id))
+
+
+@main.command()
+@click.option("--assign/--no-assign", default=True)
+@click.argument("summary")
+@click.argument("description")
+@click.pass_context
+def create(
+    ctx: click.Context,
+    assign: bool,
+    summary: str,
+    description: str,
+) -> None:
+    jira = ctx.obj["jira"]
+    fields: t.MutableMapping[str, t.Any] = {
+        "project": str(ctx.obj["project"]),
+        "issuetype": "Task",
+        "summary": summary,
+        "description": description,
+    }
+    if assign:
+        fields["assignee"] = {"name": jira.current_user()}
+    issue = jira.create_issue(fields)
+    print(issue)
+
+
+@main.command()
+@click.argument("issue_id")
+@click.argument("story_points")
+@click.pass_context
+def storypoint(
+    ctx: click.Context,
+    issue_id: str,
+    story_points: float,
+):
+    """
+    Mark issue with a certain number of storypoints.
+    """
+    jira = ctx.obj["jira"]
+    issue = jira.issue(issue_id)
+    sp_field_id = next(
+        (field["id"] for field in jira.fields() if field["name"] == "Story Points")
+    )
+    print("Story Points:", issue.get_field(sp_field_id), "->", story_points)
+    issue.update(fields={sp_field_id: float(story_points)})
+
+
+@main.command()
+@click.argument("issue_id")
+@click.pass_context
+def close(
+    ctx: click.Context,
+    issue_id: str,
+):
+    """
+    Close issue as done.
+    (ATM this is the only supported resolution.)
+    """
+    jira = ctx.obj["jira"]
+    issue = jira.issue(issue_id)
+    resolution_id = next((res.id for res in jira.resolutions() if res.name == "Done"))
+    transitions = jira.transitions(issue)
+    close_id = next((t["id"] for t in transitions if t["name"] == "Closed"))
+    jira.transition_issue(issue, close_id, resolution={"id": resolution_id})
+
+
+@main.command()
+@click.pass_context
+def types(ctx: click.Context) -> None:
+    """
+    Dump available issue types for the project.
+    """
+    jira = ctx.obj["jira"]
+    for issue_type in jira.issue_types_for_project(ctx.obj["project"]):
+        print(issue_type.name, f"(id={issue_type.id})")
+
+
+@main.command()
+@click.pass_context
+def resolutions(ctx: click.Context) -> None:
+    """
+    Dump available resolutions.
+    """
+    jira = ctx.obj["jira"]
+    for resolution in jira.resolutions():
+        print(resolution.name, f"(id={resolution.id})")
+
+
+@main.command()
+@click.pass_context
+def fields(ctx: click.Context) -> None:
+    """
+    Dump available fields.
+    """
+    jira = ctx.obj["jira"]
+    for field in jira.fields():
+        print(field["name"], "(id=" + field["id"] + ")")
+
+
+@main.command()
+@click.pass_context
+def shell(ctx: click.Context) -> None:
+    """
+    Start an interactive ipython shell.
+
+    The objects `jira` and `project` are preloaded.
+    """
+    import IPython
+
+    IPython.start_ipython(argv=[], user_ns=ctx.obj)
+
+
+if __name__ == "__main__":
+    main()