Jira API with FastAPI

jira_fastapi.py

from jira import JIRA
import uvicorn
from pydantic import BaseModel
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse

# REST API Settings
app = FastAPI(title="JiraTicketApp",
              description="Jira Ticket Server",
              version="1.0.0")
# Middleware Settings
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class CreateIssueRequestData(BaseModel):
    summary: str
    description: str
    project_key: str
    issue_type: str
    priority: str


class CommentIssueRequestData(BaseModel):
    issue_key: str
    comment: str


class GetConf:
    __URL__ = ''
    __USER__ = ''
    __API_TOKEN__ = ''


# Root API
@app.get("/")
def root() -> JSONResponse:
    return JSONResponse(status_code=200,
                        content={'message': "Jira Ticket Server"})


# Create Issue API
@app.post("/create_issue")
def create_issue(issue_data: CreateIssueRequestData) -> JSONResponse:
    """
    Create Issue API
    :param issue_data:
    :return:
    """
    data = jsonable_encoder(issue_data)
    print(data.get('summary', ''))
    print(data.get('description', ''))
    print(data.get('project_key', ''))
    print(data.get('issue_type', ''))
    print(data.get('priority', ''))  # Highest , High, Medium, Low and Lowest
    if data.get('summary', '') != '' and data.get('description',
                                                  '') != '' and data.get(
        'project_key', '') != '' and data.get('issue_type', '') != '':
        jira = JIRA(server=GetConf.__URL__,
                    basic_auth=(GetConf.__USER__, GetConf.__API_TOKEN__))
        issue_dict = {'project': {'key': data.get('project_key', '')},
                      'summary': data.get('summary'),
                      'description': data.get('description'),
                      'issuetype': {'name': data.get('issue_type', '')},
                      'priority': {'name': data.get('priority', 'Medium')}}
        new_issue = jira.create_issue(fields=issue_dict)
        return JSONResponse(status_code=200,
                            content={'message': 'new issue created',
                                     'issue_key': new_issue.key,
                                     "issue_id": new_issue.id})
    else:
        return JSONResponse(status_code=404,
                            content={'message': 'invalid data'})


# Comment Issue API
@app.post("/comment_issue")
def comment_issue(comment_data: CommentIssueRequestData) -> JSONResponse:
    """
    Comment Issue API
    :param comment_data:
    :return:
    """
    data = jsonable_encoder(comment_data)
    print(data.get('issue_key', ''))
    print(data.get('comment', ''))
    if data.get('issue_key', '') != '' and data.get('comment', '') != '':

        jira = JIRA(server=GetConf.__URL__,
                    basic_auth=(GetConf.__USER__, GetConf.__API_TOKEN__))
        jira.add_comment(data.get('issue_key', ''), data.get('comment', ''))

        return JSONResponse(status_code=200,
                            content={'message': 'new comment added'})
    else:
        return JSONResponse(status_code=404,
                            content={'message': 'invalid data'})


# Comment Issue API
@app.get("/get_projects")
def get_projects() -> JSONResponse:
    """
    Get Projects API
    :return:
    """
    try:
        jira = JIRA(server=GetConf.__URL__,
                    basic_auth=(GetConf.__USER__, GetConf.__API_TOKEN__))
        projects = jira.projects()
        pro_list = list()
        for pro in projects:
            pro_list.append(
                {"project_key": pro.key, "project_name": pro.name,
                 "project_id": pro.id})
        return JSONResponse(status_code=200,
                            content=pro_list)
    except Exception as e:
        return JSONResponse(status_code=500,
                            content={'message': 'Something went wrong'})


# Get Issues API
@app.get("/get_issues")
def get_issues(project_key: str = None) -> JSONResponse:
    """
    Get Issues API
    :param project_key:
    :return:
    """
    try:
        if project_key is not None:
            jira = JIRA(server=GetConf.__URL__,
                        basic_auth=(GetConf.__USER__, GetConf.__API_TOKEN__))
            issues_in_proj = jira.search_issues(f'project={project_key}')
            pro_list = list()
            for issue in issues_in_proj:
                create_date = \
                    str(issue.fields.created).replace("T", " ").split(".")[0]
                updated = \
                    str(issue.fields.updated).replace("T", " ").split(".")[0]
                pro_list.append(
                    {"issue_key": issue.key, "issue_id": issue.id,
                     "issue_summary": issue.fields.summary,
                     "issue_description": issue.fields.description,
                     "issue_created": create_date, "issue_updated": updated})
            return JSONResponse(status_code=200,
                                content=pro_list)
        else:
            return JSONResponse(status_code=404,
                                content={'message': 'project key is missing'})
    except Exception as e:
        print(e)
        return JSONResponse(status_code=500,
                            content={'message': 'Something went wrong'})


# Get Comments API
@app.get("/get_comments")
def get_comments(issue_key: str = None) -> JSONResponse:
    """
    Get Comments API
    :param issue_key:
    :return:
    """
    try:
        if issue_key is not None:
            jira = JIRA(server=GetConf.__URL__,
                        basic_auth=(GetConf.__USER__, GetConf.__API_TOKEN__))
            issue = jira.issue(issue_key)
            comment_list = list()
            for com in issue.fields.comment.comments:
                comment_list.append(
                    {"comment_id": com.id, "comment": com.body})
            create_date = \
                str(issue.fields.created).replace("T", " ").split(".")[0]
            updated = \
                str(issue.fields.updated).replace("T", " ").split(".")[0]
            final_obj = {"issue_key": issue.key, "issue_id": issue.id,
                         "issue_summary": issue.fields.summary,
                         "issue_description": issue.fields.description,
                         "issue_created": create_date,
                         "issue_updated": updated,
                         "comment_list": comment_list}
            return JSONResponse(status_code=200,
                                content=final_obj)
        else:
            return JSONResponse(status_code=404,
                                content={'message': 'project key is missing'})
    except Exception as e:
        return JSONResponse(status_code=500,
                            content={'message': 'Something went wrong'})


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=5000, log_level='debug')