miro_api

36class MiroApi(MiroApiExtended):
37    """
38    Exposes api endpoints as methods
39    """
40
41    def __init__(self, access_token: Union[str, Callable[[], str]]) -> None:
42        """
43        Initialize the client by passing an access token.
44
45        ```python
46        # Import the 'MiroApi' object
47        from miro_api import MiroApi
48
49        # Create a new instance of the 'MiroApi' object,
50        # and pass the OAuth access token as a parameter
51        api = MiroApi('<access_token>')
52
53        # Use the 'MiroApi' instance to send a request to the Miro REST API,
54        # and to create a new board in the team where the App is installed
55        boards = api.create_board()
56        ```
57        """
58        self.api_client = MiroApiClient(access_token)

 61class Miro:
 62    """
 63    This is the highlevel client
 64    """
 65
 66    _api_client = ApiClient()
 67    _client_id: str
 68    _client_secret: str
 69    _redirect_url: str
 70    _storage: Storage
 71
 72    def __init__(
 73        self,
 74        client_id: Optional[str] = None,
 75        client_secret: Optional[str] = None,
 76        redirect_url: Optional[str] = None,
 77        storage: Storage = InMemoryStorage(),
 78    ):
 79        """
 80        Initializes the Miro client with the given client id and client secret
 81
 82        Parameters are optional if the environment variables are defined
 83
 84            client_id: MIRO_CLIENT_ID
 85            client_secret: MIRO_CLIENT_SECRET
 86            redirect_url: MIRO_REDIRECT_URL
 87
 88        This method will raise an Exception if any of these are not defined.
 89
 90        Storage is used to persist state needed by the client (access token, refresh token).
 91        By default the client uses in memory storage for state. For production usage we recommend implementing one backed up by a permanent database.
 92        """
 93        self._client_id: str = client_id or os.environ.get("MIRO_CLIENT_ID") or ""
 94        self._client_secret: str = client_secret or os.environ.get("MIRO_CLIENT_SECRET") or ""
 95        self._redirect_url: str = redirect_url or os.environ.get("MIRO_REDIRECT_URL") or ""
 96
 97        if not self._client_id:
 98            raise Exception("miro-api: MIRO_CLIENT_ID environment variable or passing clientId is required")
 99        if not self._client_secret:
100            raise Exception("miro-api: MIRO_CLIENT_SECRET environment variable or passing clientSecret is required")
101        if not self._redirect_url:
102            raise Exception("miro-api: MIRO_REDIRECT_URL environment variable or passing redirectUrl is required")
103
104        self._storage = storage
105
106    @property
107    def api(self) -> MiroApi:
108        """Returns an instance of MiroApi using access token from given storage"""
109        return MiroApi(lambda: self.access_token)
110
111    @property
112    def is_authorized(self) -> bool:
113        """Returns True if there's a valid access token.
114        If this is False then you will need to run Miro.exchange_code_for_access_token(code) with code received through OAuth 2 flow.
115        """
116        try:
117            return bool(self.access_token)
118        except Exception:
119            return False
120
121    @property
122    def auth_url(self) -> str:
123        """Returns a URL that user should be redirected to in order to authorize the application"""
124        return self.get_auth_url()
125
126    def get_auth_url(self, state=None, team_id=None) -> str:
127        """Returns a URL that user should be redirected to in order to authorize the application, accepts an optional state argument and a teamId that will be used as a default"""
128        auth = {
129            "client_id": self._client_id,
130            "redirect_uri": self._redirect_url,
131            "response_type": "code",
132        }
133        if state:
134            auth["state"] = state
135        if team_id:
136            auth["team_id"] = team_id
137
138        return f"https://miro.com/oauth/authorize?{urllib.parse.urlencode(auth)}"
139
140    def exchange_code_for_access_token(self, url_or_code) -> str:
141        """Exchanges the authorization code for an access token by calling the token endpoint It will store the token information in storage for later reuse"""
142        code = url_or_code
143        if "?" in url_or_code:
144            url = urllib.parse.urlparse(url_or_code)
145            params = urllib.parse.parse_qs(url.query)
146            if code_param := params.get("code", [""]):
147                code = code_param[0]
148
149        if not code:
150            raise Exception("No code provided")
151
152        return self._get_token(code=code)
153
154    def revoke_token(self) -> None:
155        """Revokes currently stored access token"""
156        self.api.revoke_token(self.access_token)
157        self._storage.set(None)
158
159    def _get_token(self, code: str = "", refresh_token: str = "") -> str:
160        token_url = "https://api.miro.com/v1/oauth/token"
161        params = {
162            "client_id": self._client_id,
163            "client_secret": self._client_secret,
164        }
165        if code:
166            params.update(
167                {
168                    "code": code,
169                    "redirect_uri": self._redirect_url,
170                    "grant_type": "authorization_code",
171                }
172            )
173        elif refresh_token:
174            params.update({"refresh_token": refresh_token, "grant_type": "refresh_token"})
175
176        token_url = f"{token_url}?{urllib.parse.urlencode(params)}"
177
178        response = self._api_client.call_api("POST", token_url)
179        payload = json.loads(response.read().decode("utf-8"))
180
181        state = State(
182            payload["access_token"],
183            payload.get("refresh_token"),
184            (
185                (datetime.datetime.now() + datetime.timedelta(seconds=payload["expires_in"] - 120))
186                if payload.get("payload")
187                else None
188            ),
189        )
190
191        self._storage.set(state)
192
193        return state.access_token
194
195    @property
196    def access_token(self) -> str:
197        """Returns currently stored access token. If it has expired the client will request a new access token using the stored refresh token."""
198        state = self._storage.get()
199
200        if not state or not state.access_token:
201            raise Exception("No access token stored, run exchangeCodeForAccessToken() first")
202        if state.refresh_token and state.token_expires_at and state.token_expires_at < datetime.datetime.now():
203            return self._get_token(refresh_token=state.refresh_token)
204
205        return state.access_token