Atlassian Python API

from atlassian import (
    Jira, Confluence, Bitbucket,
    ServiceDesk, Bamboo, Xray,
    Insight, Assets, AssetsCloud,
    Portfolio, Crowd, CloudAdminOrgs, CloudAdminUsers,
    MarketPlace, StatusPage
)

from atlassian import Jira, Confluence

from atlassian import Jira, Confluence

# Initialize connections with authentication
jira = Jira(
    url="https://your-domain.atlassian.net",
    username="your-email@example.com",
    password="your-api-token"
)

confluence = Confluence(
    url="https://your-domain.atlassian.net",
    username="your-email@example.com", 
    password="your-api-token"
)

# Work with Jira issues
issue = jira.issue("PROJ-123")
new_issue = jira.create_issue(
    project="PROJ",
    summary="New feature request",
    description="Detailed description",
    issuetype="Story"
)

# Work with Confluence pages
page = confluence.get_page_by_title("Space Key", "Page Title")
confluence.create_page(
    space="SPACE",
    title="New Page",
    body="<p>Page content in storage format</p>"
)

# Work with Bitbucket repositories
bitbucket = Bitbucket(
    url="https://bitbucket.your-company.com",
    username="username",
    password="password"
)
repos = bitbucket.get_repositories()

# Work with StatusPage incidents
statuspage = StatusPage(url="https://api.statuspage.io", token="api-token")
incident = statuspage.page_create_incident(page_id, {
    "name": "Service Degradation",
    "status": "investigating",
    "impact": "minor"
})

class Jira(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def issue(self, key: str) -> dict: ...
    def create_issue(self, project: str, summary: str, description: str = "", 
                    issuetype: str = "Task", **kwargs) -> dict: ...
    def projects(self) -> List[dict]: ...
    def search_users(self, query: str) -> List[dict]: ...

class Confluence(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def get_page_by_title(self, space: str, title: str) -> dict: ...
    def create_page(self, space: str, title: str, body: str, 
                   parent_id: Optional[str] = None) -> dict: ...
    def get_all_spaces(self) -> List[dict]: ...

class Bitbucket(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def get_repositories(self) -> List[dict]: ...
    def create_repository(self, project_key: str, repository_name: str) -> dict: ...
    def get_pull_requests(self, project_key: str, repository_slug: str) -> List[dict]: ...

class ServiceDesk(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def get_service_desks(self) -> List[dict]: ...
    def create_customer_request(self, service_desk_id: str, request_type_id: str,
                               summary: str, description: str = "") -> dict: ...
    def get_customer_requests(self, service_desk_id: str) -> List[dict]: ...

class Bamboo(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def get_builds(self) -> List[dict]: ...
    def start_build(self, plan_key: str) -> dict: ...
    def get_results(self, plan_key: str) -> List[dict]: ...

class Xray(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def get_tests(self, project_key: str) -> List[dict]: ...
    def get_test_runs(self, test_key: str) -> List[dict]: ...
    def import_execution_results(self, execution_results: dict) -> dict: ...

class Insight(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def get_object(self, object_id: str) -> dict: ...
    def create_object(self, object_type_id: str, attributes: dict) -> dict: ...
    def get_object_schemas(self) -> List[dict]: ...

class AssetsCloud(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def get_workspaces(self) -> List[dict]: ...
    def get_objects(self, workspace_id: str) -> List[dict]: ...

class Portfolio(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def get_plans(self) -> List[dict]: ...
    def create_plan(self, name: str, description: str = "") -> dict: ...

class StatusPage(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def page_list_pages(self) -> List[dict]: ...
    def page_create_incident(self, page_id: str, incident: dict) -> dict: ...
    def page_get_components(self, page_id: str) -> dict: ...
    def page_create_subscriber(self, page_id: str, subscriber: dict) -> dict: ...
    def page_add_data_to_metric(self, page_id: str, metric_id: str, data: dict) -> dict: ...

class Crowd(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def get_users(self) -> List[dict]: ...
    def create_user(self, username: str, email: str, first_name: str, 
                   last_name: str) -> dict: ...
    def authenticate_user(self, username: str, password: str) -> bool: ...

class CloudAdminOrgs(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def get_orgs(self) -> List[dict]: ...
    def get_org_users(self, org_id: str) -> List[dict]: ...

class MarketPlace(AtlassianRestAPI):
    def __init__(self, url: str, *args, **kwargs): ...
    def get_addon_info(self, addon_key: str) -> dict: ...

# API Token (recommended for Cloud)
client = Jira(
    url="https://your-domain.atlassian.net",
    username="email@example.com",
    password="api-token"
)

# OAuth 1.0a
client = Jira(
    url="https://your-domain.atlassian.net",
    oauth={
        "access_token": "token",
        "access_token_secret": "secret",
        "consumer_key": "key",
        "key_cert": "private_key"
    }
)

# Basic Authentication (Server instances)
client = Jira(
    url="https://jira.company.com",
    username="username",
    password="password"
)

from atlassian.errors import (
    ApiError, ApiNotFoundError, ApiPermissionError,
    ApiValueError, ApiConflictError, ApiNotAcceptable,
    JsonRPCError, JsonRPCRestrictionsError
)

class ApiError(Exception):
    """Base exception for all API errors with reason attribute"""
    def __init__(self, *args, reason=None): ...

class ApiNotFoundError(ApiError):
    """HTTP 404 - Resource not found"""
    pass

class ApiPermissionError(ApiError):
    """HTTP 403 - Permission denied or authentication failed"""
    pass

class ApiValueError(ApiError):
    """HTTP 400 - Invalid request parameters or malformed data"""
    pass

class ApiConflictError(ApiError):
    """HTTP 409 - Resource conflict (e.g., duplicate names)"""
    pass

class ApiNotAcceptable(ApiError):
    """HTTP 406 - Request format not acceptable"""
    pass

class JsonRPCError(Exception):
    """Base exception for JSON-RPC specific errors"""
    pass

class JsonRPCRestrictionsError(JsonRPCError):
    """JSON-RPC permission or access restrictions"""
    pass

try:
    issue = jira.issue("INVALID-123")
except ApiNotFoundError:
    print("Issue not found")
except ApiPermissionError:
    print("Permission denied")
except ApiValueError as e:
    print(f"Invalid request: {e}")
except ApiError as e:
    print(f"API error {e.reason}: {e}")

class AtlassianRestAPI:
    def __init__(self, url: str, username: str = None, password: str = None,
                 timeout: int = 75, api_root: str = "rest/api", 
                 api_version: str = "latest", verify_ssl: bool = True,
                 session: Optional[requests.Session] = None,
                 oauth: Optional[dict] = None, oauth2: Optional[dict] = None,
                 cloud: bool = False, token: Optional[str] = None,
                 backoff_and_retry: bool = False, **kwargs):
        """
        Initialize Atlassian REST API client.
        
        Parameters:
        - url: Base URL of Atlassian instance
        - username: Username for authentication
        - password: Password or API token for authentication
        - timeout: Request timeout in seconds
        - api_root: API root path (default "rest/api")
        - api_version: API version ("latest" or specific version)
        - verify_ssl: Verify SSL certificates
        - session: Custom requests session
        - oauth: OAuth 1.0a configuration dict
        - oauth2: OAuth 2.0 configuration dict
        - cloud: True for Cloud, False for Server/Data Center
        - token: Bearer token for authentication
        - backoff_and_retry: Enable automatic retry with backoff
        """

    def get(self, path: str, **kwargs) -> Union[dict, str, bytes]:
        """Execute GET request"""
        
    def post(self, path: str, data: dict = None, **kwargs) -> Union[dict, str]:
        """Execute POST request"""
        
    def put(self, path: str, data: dict = None, **kwargs) -> Union[dict, str]:
        """Execute PUT request"""
        
    def delete(self, path: str, **kwargs) -> Union[dict, str]:
        """Execute DELETE request"""

from atlassian.typehints import T_id, T_resp_json
from typing import Union, Optional, List, Dict, Any

# Resource identifier type
T_id = Union[str, int]

# JSON response type  
T_resp_json = Union[dict, None]

# Common parameter types
IssueFields = Dict[str, Any]
JQLQuery = str
ProjectKey = str
IssueKey = str

# Get all issues with pagination
all_issues = []
start_at = 0
max_results = 50

while True:
    results = jira.search_issues(
        jql="project = PROJ",
        start=start_at,
        limit=max_results
    )
    
    all_issues.extend(results["issues"])
    
    if len(results["issues"]) < max_results:
        break
        
    start_at += max_results

# Bulk create issues
issues_data = [
    {"summary": "Task 1", "project": "PROJ", "issuetype": "Task"},
    {"summary": "Task 2", "project": "PROJ", "issuetype": "Task"},
]

for issue_data in issues_data:
    jira.create_issue(**issue_data)

from atlassian.errors import ApiError
import time

def retry_api_call(func, *args, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except ApiError as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff