Composio SDK client.

1
@staticmethod
2
def get_latest() -> 'Composio':
3
    """Get latest composio client from the runtime stack."""
4
    if len(_clients) == 0:
5
        _ = Composio()
6
    return _clients[-1]

1
@property
2
def api_key(self) -> str:
3
    if self._api_key is None:
4
        user_data_path = LOCAL_CACHE_DIRECTORY / USER_DATA_FILE_NAME
5
        user_data = UserData.load(path=user_data_path) if user_data_path.exists() else None
6
        env_api_key = user_data.api_key if user_data is not None and user_data.api_key is not None else os.environ.get(ENV_COMPOSIO_API_KEY)
7
        if env_api_key:
8
            self._api_key = env_api_key
9
    if self._api_key is None:
10
        raise ApiKeyNotProvidedError
11
    self._api_key = self.validate_api_key(key=t.cast(str, self._api_key), base_url=self.base_url)
12
    return self._api_key

1
@api_key.setter
2
def api_key(self, value: str) -> None:
3
    self._api_key = value

1
@property
2
def http(self) -> HttpClient:
3
    if not self._http:
4
        self._http = HttpClient(base_url=self.base_url, api_key=self.api_key, runtime=self.runtime)
5
    return self._http

1
@http.setter
2
def http(self, value: HttpClient) -> None:
3
    self._http = value

1
@property
2
def long_timeout_http(self) -> HttpClient:
3
    if not self._long_timeout_http:
4
        self._long_timeout_http = HttpClient(base_url=self.base_url, api_key=self.api_key, runtime=self.runtime, timeout=180.0)
5
    return self._long_timeout_http

1
@long_timeout_http.setter
2
def long_timeout_http(self, value: HttpClient) -> None:
3
    self._long_timeout_http = value

1
@staticmethod
2
def validate_api_key(key: str, base_url: t.Optional[str]=None) -> str:
3
    """Validate given API key."""
4
    if key in _valid_keys:
5
        return key
6
    base_url = base_url or get_api_url_base()
7
    response = requests.get(url=base_url + str(v1 / 'client' / 'auth' / 'client_info'), headers={'x-api-key': key, 'x-request-id': generate_request_id()}, timeout=60)
8
    if response.status_code in (401, 403):
9
        raise ApiKeyError('API Key is not valid!')
10
    if response.status_code != 200:
11
        raise ApiKeyError(f'Unexpected error: HTTP {response.status_code}')
12
    _valid_keys.add(key)
13
    return key

1
@staticmethod
2
def generate_auth_key(base_url: t.Optional[str]=None) -> str:
3
    """Generate auth key."""
4
    http = HttpClient(base_url=base_url or get_api_url_base(), api_key='')
5
    response = http.get(url=str(v1.cli.generate_cli_session))
6
    if response.status_code != 200:
7
        raise HTTPError(message=response.content.decode(), status_code=response.status_code)
8
    data = response.json()
9
    return data['key']

1
@staticmethod
2
def validate_auth_session(key: str, code: str, base_url: t.Optional[str]=None) -> str:
3
    """
4
        Validate API session.
5
6
        :param key: Session key
7
        :param code: Authentication code
8
        """
9
    http = HttpClient(base_url=base_url or get_api_url_base(), api_key='')
10
    response = http.get(str(v1.cli.verify_cli_code({'key': key, 'code': code})))
11
    if response.status_code != 200:
12
        raise HTTPError(message=response.content.decode(), status_code=response.status_code)
13
    data = response.json()
14
    return data['apiKey']

1
def get_entity(self, id: str=DEFAULT_ENTITY_ID) -> 'Entity':
2
    """
3
        Create Entity object.
4
5
        :param id: Entity ID
6
        :return: Entity object.
7
        """
8
    return Entity(id=id, client=self)

1
@deprecated(version='0.5.52', replacement='execute_action')
2
def execute(self, action: Action, params: t.Dict, connected_account_id: t.Optional[str]=None, session_id: t.Optional[str]=None, text: t.Optional[str]=None, auth: t.Optional[CustomAuthObject]=None) -> t.Dict:
3
    """
4
        Execute an action.
5
6
        :param action: Action ID (Enum)
7
        :param params: Parameters for executing actions
8
        :param connected_account_id: Connection ID if you want to use a specific
9
                connection
10
        :param session_id: ID of the current workspace session
11
        :return: Dictionary containing execution result
12
        """
13
    return self._execute(action, params, connected_account_id, session_id, text, auth)

1
def get_connection(self, app: t.Optional[AppType]=None, connected_account_id: t.Optional[str]=None) -> ConnectedAccountModel:
2
    """
3
        Get connected account for an action.
4
5
        :param app: App name
6
        :param connected_account_id: Connected account ID to use as filter
7
        :return: Connected account object
8
        :raises: If no connected account found for given entity ID
9
        """
10
    if connected_account_id is not None:
11
        return self.client.connected_accounts.get(connection_id=connected_account_id)
12
    latest_account = None
13
    latest_creation_date = datetime.fromtimestamp(0.0)
14
    connected_accounts = self.client.connected_accounts.get(entity_ids=[self.id], active=True)
15
    app = str(app).lower()
16
    for connected_account in connected_accounts:
17
        if app == connected_account.appUniqueId:
18
            creation_date = datetime.fromisoformat(connected_account.createdAt.replace('Z', '+00:00'))
19
            if latest_account is None or creation_date < latest_creation_date:
20
                latest_creation_date = creation_date
21
                latest_account = connected_account
22
    if latest_account is None:
23
        entity = self.id
24
        suggestion = f'composio add {app}' if entity == DEFAULT_ENTITY_ID else f'composio add {app} -e {entity}'
25
        note = f'Run this command to create a new connection: {suggestion}'
26
        doc_note = 'Read more here: https://dub.composio.dev/auth-help'
27
        if sys.version_info >= (3, 11):
28
            exception = NoItemsFound(f'Could not find a connection with app={app!r}, connected_account_id={connected_account_id!r} and entity={entity!r}.')
29
            exception.add_note(note)
30
            exception.add_note(doc_note)
31
        else:
32
            exception = NoItemsFound(f'Could not find a connection with app={app!r}, connected_account_id={connected_account_id!r} and entity={entity!r}.\n{note}\n{doc_note}')
33
        raise exception
34
    return latest_account

1
def get_connections(self) -> t.List[ConnectedAccountModel]:
2
    """
3
        Get all connections for an entity.
4
        """
5
    return self.client.connected_accounts.get(entity_ids=[self.id], active=True)

1
def enable_trigger(self, app: t.Union[str, App], trigger_name: str, config: t.Dict[str, t.Any]) -> t.Dict:
2
    """
3
        Enable a trigger for an entity.
4
5
        :param app: App name
6
        :param trigger_name: Trigger name
7
        :param config: Trigger config
8
        """
9
    connected_account = self.get_connection(app=app)
10
    return self.client.triggers.enable(name=trigger_name, connected_account_id=connected_account.id, config=config)

1
def disable_trigger(self, trigger_id: str) -> t.Dict:
2
    """
3
        Disable a trigger for an entity.
4
5
        :param trigger_id: Trigger ID
6
        """
7
    return self.client.triggers.disable(id=trigger_id)

1
def get_active_triggers(self) -> t.List[ActiveTriggerModel]:
2
    """
3
        Get all active triggers for an entity.
4
        """
5
    connected_accounts = self.get_connections()
6
    return self.client.active_triggers.get(connected_account_ids=[connected_account.id for connected_account in connected_accounts])

1
def initiate_connection(self, app_name: t.Union[str, App], auth_mode: t.Optional[str]=None, auth_config: t.Optional[t.Dict[str, t.Any]]=None, redirect_url: t.Optional[str]=None, integration: t.Optional[IntegrationModel]=None, use_composio_auth: bool=True, force_new_integration: bool=False, connected_account_params: t.Optional[t.Dict]=None, labels: t.Optional[t.List]=None) -> ConnectionRequestModel:
2
    """
3
        Initiate an integration connection process for a specified application.
4
5
        :param app_name: The name of the application or an App enum instance.
6
        :param auth_mode: Optional authentication mode to be used.
7
        :param auth_config: Optional dictionary containing authentication configuration details.
8
        :param redirect_url: Optional URL to which a user will be redirected after authentication.
9
        :param integration: Optional existing IntegrationModel instance to be used.
10
        :return: A ConnectionRequestModel instance representing the initiated connection.
11
        """
12
    app = self.client.apps.get(name=App(app_name).slug)
13
    timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
14
    if integration is None and auth_mode is not None:
15
        if auth_mode not in AUTH_SCHEME_WITH_INITIATE:
16
            raise InvalidParams(f"'auth_mode' should be one of {AUTH_SCHEME_WITH_INITIATE}")
17
        auth_mode = t.cast(AuthSchemeType, auth_mode)
18
        if 'OAUTH' not in auth_mode:
19
            use_composio_auth = False
20
        integration = self.client.integrations.create(app_id=app.appId, name=f'{app_name}_{timestamp}', auth_mode=auth_mode, auth_config=auth_config, use_composio_auth=use_composio_auth, force_new_integration=force_new_integration)
21
    if integration is None and auth_mode is None:
22
        integration = self.client.integrations.create(app_id=app.appId, auth_config=auth_config, name=f'{app_name}_{timestamp}', use_composio_auth=use_composio_auth, force_new_integration=force_new_integration)
23
    return self.client.connected_accounts.initiate(integration_id=t.cast(IntegrationModel, integration).id, entity_id=self.id, params=connected_account_params, labels=labels, redirect_url=redirect_url)