Add pj

.gitignore

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

pj/.gitignore

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

pj/pj.py

@@ -0,0 +1,140 @@
+#!/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 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()
+@click.argument("search_phrase")
+@click.pass_context
+def search(ctx: click.Context, search_phrase) -> 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.argument("issue_id")
+@click.pass_context
+def show(ctx: click.Context, issue_id: str) -> None:
+    jira = ctx.obj["jira"]
+    issue = jira.issue(issue_id)
+    print(json.dumps(issue.raw))
+
+
+@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.pass_context
+def types(ctx: click.Context) -> None:
+    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 shell(ctx: click.Context) -> None:
+    import IPython
+
+    IPython.start_ipython(argv=[], user_ns=ctx.obj)
+
+
+if __name__ == "__main__":
+    main()