quick-start-connectors

1
import requests
2
from flask import current_app as app
3

4
from . import UpstreamProviderError
5

6
client = None
7

8

9
class AhaApiClient:
10
    ALLOWED_ENTITY_TYPES = {
11
        "users": {
12
            "search_fields": ["name", "email"],
13
            "mapping": {"name": "text", "email": "title"},
14
        },
15
        "capacity_scenarios": {
16
            "search_fields": ["name"],
17
            "mapping": {"name": "text"},
18
        },
19
        "epics": {
20
            "search_fields": ["q"],
21
            "mapping": {"name": "text", "reference_num": "title"},
22
        },
23
        "features": {
24
            "search_fields": ["q"],
25
            "mapping": {"name": "text", "reference_num": "title"},
26
        },
27
        "goals": {
28
            "search_fields": ["name", "reference_num", "description"],
29
            "mapping": {"description": "text", "name": "title"},
30
        },
31
        "ideas": {
32
            "search_fields": ["q"],
33
            "mapping": {"description": "text", "name": "title"},
34
        },
35
        "initiatives": {
36
            "search_fields": ["q"],
37
            "mapping": {"description": "text", "name": "title"},
38
        },
39
        "integrations": {"search_fields": ["name"], "mapping": {"name": "text"}},
40
        "products": {
41
            "search_fields": ["name"],
42
            "mapping": {"name": "text", "reference_prefix": "title"},
43
        },
44
        "release_phases": {
45
            "search_fields": ["name", "description"],
46
            "mapping": {"description": "text", "name": "title"},
47
        },
48
        "strategy_models": {
49
            "search_fields": ["name", "kind"],
50
            "mapping": {"name": "text", "kind": "title"},
51
        },
52
        "strategy_positions": {
53
            "search_fields": ["name", "kind"],
54
            "mapping": {"name": "text", "kind": "title"},
55
        },
56
        "strategy_visions": {
57
            "search_fields": ["name", "description"],
58
            "mapping": {"description": "text", "name": "title"},
59
        },
60
        "teams": {"search_fields": ["name"], "mapping": {"name": "text"}},
61
        "tasks": {"search_fields": ["name"], "mapping": {"name": "text"}},
62
    }
63

64
    def __init__(self, domain, api_key, allowed_entities, search_limit):
65
        self.api_url = f"https://{domain}.aha.io/api/v1/"
66
        self.headers = {"Authorization": f"Bearer {api_key}"}
67
        self.allowed_entities = allowed_entities
68
        self.search_limit = search_limit
69

70
    def get_allowed_entities(self):
71
        return self.allowed_entities
72

73
    def get_allowed_entity_types(self):
74
        return self.ALLOWED_ENTITY_TYPES
75

76
    def get_search_limit(self):
77
        return self.search_limit
78

79
    def get(self, url, params={}):
80
        response = requests.get(url, headers=self.headers, params=params)
81

82
        if response.status_code != 200:
83
            message = response.text or f"Error: HTTP {response.status_code}"
84
            raise UpstreamProviderError(message)
85

86
        return response.json()
87

88
    def get_entities_by_type(self, entity_type, params={}):
89
        url = f"{self.api_url}/{entity_type}"
90
        return self.get(url, params)
91

92

93
def get_client():
94
    global client
95
    assert (domain := app.config.get("DOMAIN")), "AHA_DOMAIN must be set"
96
    assert (api_key := app.config.get("API_KEY")), "AHA_API_KEY must be set"
97
    search_limit = app.config.get("SEARCH_LIMIT", 20)
98
    allowed_entities = app.config.get(
99
        "ALLOWED_ENTITIES",
100
        [
101
            "users",
102
            "capacity_scenarios",
103
            "epics",
104
            "features",
105
            "goals",
106
            "ideas",
107
            "initiatives",
108
            "integrations",
109
            "products",
110
            "release_phases",
111
            "strategy_models",
112
            "strategy_positions",
113
            "strategy_visions",
114
            "teams",
115
            "tasks",
116
        ],
117
    )
118

119
    if not client:
120
        client = AhaApiClient(domain, api_key, allowed_entities, search_limit)
121

122
    return client
123