Composio server object collections

1
def to_trigger_names(
2
    triggers: t.Union[t.List[str], t.List[Trigger], t.List[TriggerType]],
3
) -> str:
4
    """Get trigger names as a string."""
5
    return ",".join([Trigger(trigger).slug for trigger in triggers])

1
def save_user_access_data(self, client: 'Composio', field_inputs: t.Dict, redirect_url: t.Optional[str]=None, entity_id: t.Optional[str]=None) -> t.Dict:
2
    """Save user access data."""
3
    connected_account = client.connected_accounts.get(connection_id=self.connectedAccountId)
4
    resp = client.http.post(url=str(v1 / 'connectedAccounts'), json={'integrationId': connected_account.integrationId, 'data': field_inputs, 'redirectUri': redirect_url, 'userUuid': entity_id})
5
    return resp.json()

1
def wait_until_active(self, client: 'Composio', timeout: float=60.0) -> 'ConnectedAccountModel':
2
    start_time = time.time()
3
    while time.time() - start_time < timeout:
4
        connection = client.connected_accounts.get(connection_id=self.connectedAccountId)
5
        if connection.status == 'ACTIVE':
6
            return connection
7
        time.sleep(1)
8
    raise SDKTimeoutError('Connection did not become active within the timeout period.')

1
@t.overload
2
def get(self) -> t.List[ConnectedAccountModel]:
3
    """
4
        Get all connected accounts
5
6
        :return: List of Connected accounts
7
        """

1
@t.overload
2
def get(self, connection_id: str) -> ConnectedAccountModel:
3
    """
4
        Get an account by connection ID
5
6
        :param connection_id: ID of the connection to filter by
7
        :return: Connected account
8
        """

1
@t.overload
2
def get(self, *, entity_ids: t.Optional[t.Sequence[str]]=None, active: bool=False) -> t.List[ConnectedAccountModel]:
3
    """
4
        Get a list of connected accounts by entity IDs
5
6
        :param entity_ids: List of entity IDs to filter by
7
        :param active: Returns account which are currently active
8
        :return: List of connected accounts
9
        """

1
def get(self, connection_id: t.Optional[str]=None, *, entity_ids: t.Optional[t.Sequence[str]]=None, active: bool=False) -> t.Union[ConnectedAccountModel, t.List[ConnectedAccountModel]]:
2
    """
3
        Get a list of connected accounts.
4
5
        :param entity_ids: List of entity IDs to filter by
6
        :param connection_id: Return the connected account by a specific
7
                connection ID
8
        :param active: Returns account which are currently active
9
        :return: List of connected accounts
10
        """
11
    entity_ids = entity_ids or ()
12
    if connection_id is not None and len(entity_ids) > 0:
13
        raise InvalidParams(message='Cannot use both `connection_id` and `entity_ids` parameters as filter')
14
    if connection_id is not None:
15
        response = self._raise_if_required(self.client.http.get(url=str(self.endpoint / connection_id)))
16
        return self.model(**response.json())
17
    queries = {'pageSize': '99999999'}
18
    if len(entity_ids) > 0:
19
        queries['user_uuid'] = ','.join(entity_ids)
20
    if active:
21
        queries['showActiveOnly'] = 'true'
22
    response = self._raise_if_required(self.client.http.get(url=str(self.endpoint(queries=queries))))
23
    return [self.model(**account) for account in response.json().get('items', [])]

1
def initiate(self, integration_id: str, entity_id: t.Optional[str]=None, params: t.Optional[t.Dict]=None, labels: t.Optional[t.List]=None, redirect_url: t.Optional[str]=None) -> ConnectionRequestModel:
2
    """Initiate a new connected account."""
3
    response = self._raise_if_required(response=self.client.http.post(url=str(self.endpoint), json={'integrationId': integration_id, 'userUuid': entity_id, 'data': params or {}, 'labels': labels or [], 'redirectUri': redirect_url}))
4
    return ConnectionRequestModel(**response.json())

1
def info(self, connection_id: str) -> ConnectionParams:
2
    response = self._raise_if_required(self.client.http.get(url=str(self.endpoint / connection_id / 'info')))
3
    return ConnectionParams(**response.json())

1
@t.overload
2
def get(self) -> t.List[AppModel]:
3
    """Get available apps."""

1
@t.overload
2
def get(self, name: t.Optional[str]=None) -> AppModel:
3
    """Get a specific app."""

1
def get(self, name: t.Optional[str]=None) -> t.Union[AppModel, t.List[AppModel]]:
2
    """Get apps."""
3
    if name is not None:
4
        return self.model(**self._raise_if_required(response=self.client.http.get(url=str(self.endpoint / name))).json())
5
    return super().get(queries={})

1
def set(self, url: str) -> CallbackModel:
2
    """Set callback URL."""
3
    response = self._raise_if_required(response=self.client.http.post(url=str(self.endpoint / 'setCallbackURL'), json={'callbackURL': url}))
4
    return response.json()

1
def get(self) -> str:
2
    """Get current callback URL."""
3
    response = self._raise_if_required(response=self.client.http.get(url=str(self.endpoint / 'callback_url')))
4
    return response.json().get('callbackURL')

1
def validate_filters(self, filters: _TriggerEventFilters):
2
    docs_link_msg = '\nRead more here: https://docs.composio.dev/introduction/intro/quickstart_3'
3
    if not isinstance(filters, dict):
4
        raise InvalidParams('Expected filters to be a dictionary' + docs_link_msg)
5
    expected_filters = list(_TriggerEventFilters.__annotations__)
6
    for filter, value in filters.items():
7
        if filter not in expected_filters:
8
            error_msg = f'Unexpected filter {filter!r}'
9
            possible_values = difflib.get_close_matches(filter, expected_filters, n=1)
10
            if possible_values:
11
                possible_value, = possible_values
12
                error_msg += f' Did you mean {possible_value!r}?'
13
            raise InvalidTriggerFilters(error_msg + docs_link_msg)
14
        if filter == 'app_name':
15
            if isinstance(value, App):
16
                slug = value.slug
17
            elif isinstance(value, str):
18
                slug = value
19
            else:
20
                raise InvalidTriggerFilters(f"Expected 'app_name' to be App or str, found {value!r}" + docs_link_msg)
21
            slug = slug.upper()
22
            app_names = list(App.iter())
23
            if slug not in app_names:
24
                error_msg = f'App {slug!r} does not exist.'
25
                possible_values = difflib.get_close_matches(slug, app_names, n=1)
26
                if possible_values:
27
                    possible_value, = possible_values
28
                    error_msg += f' Did you mean {possible_value!r}?'
29
                raise InvalidTriggerFilters(error_msg + docs_link_msg)
30
            active_triggers = [trigger.triggerName for trigger in self.client.active_triggers.get()]
31
            apps_for_triggers = {Trigger(trigger).app.upper() for trigger in active_triggers}
32
            if slug not in apps_for_triggers:
33
                error_msg = f'App {slug!r} has no triggers enabled on your account.\nFind the possible triggers by running `composio triggers`.'
34
                raise InvalidTriggerFilters(error_msg + docs_link_msg)
35
        if filter == 'trigger_name':
36
            if isinstance(value, Trigger):
37
                slug = value.slug
38
            elif isinstance(value, str):
39
                slug = value
40
            else:
41
                raise InvalidTriggerFilters(f"Expected 'trigger_name' to be Trigger or str, found {value!r}" + docs_link_msg)
42
            slug = slug.upper()
43
            trigger_names = list(Trigger.iter())
44
            if slug not in trigger_names:
45
                error_msg = f'Trigger {slug!r} does not exist.'
46
                possible_values = difflib.get_close_matches(slug, trigger_names, n=1)
47
                if possible_values:
48
                    possible_value, = possible_values
49
                    error_msg += f' Did you mean {possible_value!r}?'
50
                raise InvalidTriggerFilters(error_msg + docs_link_msg)
51
            active_triggers = [trigger.triggerName for trigger in self.client.active_triggers.get()]
52
            if slug not in active_triggers:
53
                error_msg = f'Trigger {slug!r} is not enabled on your account.\nEnable the trigger by doing `composio triggers enable {slug}`.'
54
                raise InvalidTriggerFilters(error_msg + docs_link_msg)

1
def callback(self, filters: t.Optional[_TriggerEventFilters]=None) -> t.Callable[[TriggerCallback], TriggerCallback]:
2
    """Register a trigger callaback."""
3
    if filters is not None:
4
        self.validate_filters(filters)
5
6
    def _wrap(f: TriggerCallback) -> TriggerCallback:
7
        self._callbacks.append((f, filters or {}))
8
        return f
9
    return _wrap

1
def handle_event(self, event: str) -> None:
2
    """Filter events and call the callback function."""
3
    data = self._parse_payload(event=event)
4
    if data is None:
5
        self.logger.error(f'Error parsing trigger payload: {event}')
6
        return
7
    self.logger.debug(f'Received trigger event with trigger ID: {data.metadata.id} and trigger name: {data.metadata.triggerName}')
8
    awaitables: t.List[Future] = []
9
    with ThreadPoolExecutor() as executor:
10
        for callback, filters in self._callbacks:
11
            awaitables.append(executor.submit(self._handle_callback, callback, data, filters))
12
    _ = [future.result() for future in awaitables]

1
def handle_chunked_events(self, event: str) -> None:
2
    """Handle chunked events."""
3
    data = _ChunkedTriggerEventData(**json.loads(event))
4
    if data.id not in self._chunks:
5
        self._chunks[data.id] = {}
6
    self._chunks[data.id][data.index] = data.chunk
7
    if data.final:
8
        _chunks = self._chunks.pop(data.id)
9
        self.handle_event(event=''.join([_chunks[idx] for idx in sorted(_chunks)]))

1
def is_alive(self) -> bool:
2
    """Check if subscription is live."""
3
    return self._alive

1
def has_errored(self) -> bool:
2
    """Check if the connection errored and disconnected."""
3
    return self._connection.socket is None or self._connection.socket.has_errored

1
def set_alive(self) -> None:
2
    """Set `_alive` to True."""
3
    self._alive = True

1
@te.deprecated('Use `wait_forever` instead')
2
def listen(self) -> None:
3
    """Wait infinitely."""
4
    self.wait_forever()

1
def wait_forever(self) -> None:
2
    """Wait infinitely."""
3
    while self.is_alive() and (not self.has_errored()):
4
        time.sleep(1)

1
def stop(self) -> None:
2
    """Stop the trigger listener."""
3
    self._connection.disconnect()
4
    self._alive = False

1
def restart(self) -> None:
2
    """Restart the subscription connection"""
3
    self._connection.disconnect()
4
    self._connection._connect()

1
def get(self, trigger_names: t.Optional[t.List[TriggerType]]=None, apps: t.Optional[t.List[str]]=None) -> t.List[TriggerModel]:
2
    """
3
        List active triggers
4
5
        :param trigger_names: Trigger names to filter by, can be a list of strings or Trigger objects
6
        :param app_names: App names to filter by
7
        :return: List of triggers filtered by provided parameters
8
        """
9
    queries = {}
10
    if trigger_names is not None and len(trigger_names) > 0:
11
        queries['triggerIds'] = to_trigger_names(trigger_names)
12
    if apps is not None and len(apps) > 0:
13
        queries['appNames'] = ','.join(apps)
14
    return super().get(queries=queries)

1
def enable(self, name: str, connected_account_id: str, config: t.Dict[str, t.Any]) -> t.Dict:
2
    """
3
        Enable a trigger
4
5
        :param name: Name of the trigger
6
        :param connected_account_id: ID of the relevant connected account
7
        """
8
    response = self._raise_if_required(self.client.http.post(url=str(self.endpoint.enable / connected_account_id / name), json={'triggerConfig': config}))
9
    return response.json()

1
def disable(self, id: str) -> t.Dict:
2
    """
3
        Disable a trigger
4
5
        :param name: Name of the trigger
6
        :param connected_account_id: ID of the relevant connected account
7
        """
8
    response = self._raise_if_required(self.client.http.patch(url=str(self.endpoint / 'instance' / id / 'status'), json={'enabled': False}))
9
    return response.json()

1
def delete(self, id: str) -> t.Dict:
2
    """
3
        Delete a trigger
4
5
        :param id: ID of the trigger to be deleted
6
        """
7
    response = self._raise_if_required(self.client.http.delete(url=str(self.endpoint / 'instance' / id)))
8
    return response.json()

1
def subscribe(self, timeout: float=15.0) -> TriggerSubscription:
2
    """Subscribe to a trigger and receive trigger events."""
3
    self.logger.debug('Creating trigger subscription')
4
    response = self._raise_if_required(response=self.client.http.get(url='/v1/client/auth/client_info'))
5
    client_id = response.json().get('client', {}).get('id')
6
    if client_id is None:
7
        raise TriggerSubscriptionError('Error fetching client ID')
8
    pusher = _PusherClient(client_id=client_id, client=self.client)
9
    return pusher.connect(timeout=timeout)

1
def get(self, trigger_ids: t.Optional[t.List[str]]=None, connected_account_ids: t.Optional[t.List[str]]=None, integration_ids: t.Optional[t.List[str]]=None, trigger_names: t.Optional[t.List[t.Union[str, Trigger]]]=None) -> t.List[ActiveTriggerModel]:
2
    """List active triggers."""
3
    trigger_ids = trigger_ids or []
4
    connected_account_ids = connected_account_ids or []
5
    integration_ids = integration_ids or []
6
    trigger_names = trigger_names or []
7
    queries = {}
8
    if len(trigger_ids) > 0:
9
        queries['triggerIds'] = ','.join(trigger_ids)
10
    if len(connected_account_ids) > 0:
11
        queries['connectedAccountIds'] = ','.join(connected_account_ids)
12
    if len(integration_ids) > 0:
13
        queries['integrationIds'] = ','.join(integration_ids)
14
    if len(trigger_names) > 0:
15
        queries['triggerNames'] = to_trigger_names(trigger_names)
16
    return super().get(queries=queries)

1
@t.overload
2
def get(self) -> t.List[ActionModel]:
3
    ...

1
@t.overload
2
def get(self, action: t.Optional[ActionType]=None) -> ActionModel:
3
    ...

1
@t.overload
2
def get(self, actions: t.Optional[t.Sequence[ActionType]]=None, apps: t.Optional[t.Sequence[AppType]]=None, tags: t.Optional[t.Sequence[TagType]]=None, limit: t.Optional[int]=None, use_case: t.Optional[str]=None, allow_all: bool=False) -> t.List[ActionModel]:
3
    ...

1
def get(self, action: t.Optional[ActionType]=None, *, actions: t.Optional[t.Sequence[ActionType]]=None, apps: t.Optional[t.Sequence[AppType]]=None, tags: t.Optional[t.Sequence[TagType]]=None, limit: t.Optional[int]=None, use_case: t.Optional[str]=None, allow_all: bool=False) -> t.Union[ActionModel, t.List[ActionModel]]:
2
    """
3
        Get a list of apps by the specified filters.
4
5
        :param actions: Filter by the list of Actions.
6
        :param action: Get data for this action.
7
        :param apps: Filter by the list of Apps.
8
        :param tags: Filter by the list of given Tags.
9
        :param limit: Limit the number of actions to a specific number.
10
        :param use_case: Filter by use case.
11
        :param allow_all: Allow querying all of the actions for a specific
12
                        app
13
        :return: List of actions
14
        """
15
    if action is not None:
16
        return self._get_action(action=action)
17
    return self._get_actions(actions=actions, apps=apps, tags=tags, limit=limit, use_case=use_case, allow_all=allow_all)

1
def execute(self, action: Action, params: t.Dict, entity_id: str='default', connected_account: t.Optional[str]=None, session_id: t.Optional[str]=None, text: t.Optional[str]=None, auth: t.Optional[CustomAuthObject]=None, allow_tracing: bool=False) -> t.Dict:
2
    """
3
        Execute an action on the specified entity with optional connected account.
4
5
        :param action: The Action object to be executed.
6
        :param params: A dictionary of parameters to be passed to the action.
7
        :param entity_id: The unique identifier of the entity on which the action is executed.
8
        :param connected_account: Optional connected account ID if required for the action.
9
        :param session_id: ID of the current workspace session
10
        :return: A dictionary containing the response from the executed action.
11
        """
12
    if action.no_auth:
13
        return self._raise_if_required(self.client.long_timeout_http.post(url=str(self.endpoint / action.slug / 'execute'), json={'appName': action.app, 'input': params, 'text': text, 'version': action.version, 'sessionInfo': {'sessionId': session_id}, 'allowTracing': allow_tracing})).json()
14
    if connected_account is None and auth is None:
15
        raise InvalidParams('`connected_account` cannot be `None` when executing an app which requires authentication')
16
    return self._raise_if_required(self.client.long_timeout_http.post(url=str(self.endpoint / action.slug / 'execute'), json={'connectedAccountId': connected_account, 'entityId': entity_id, 'appName': action.app, 'input': params, 'text': text, 'version': action.version, 'authConfig': self._serialize_auth(auth=auth), 'sessionInfo': {'sessionId': session_id}, 'allowTracing': allow_tracing})).json()

1
def request(self, connection_id: str, endpoint: str, method: str, body: t.Optional[t.Dict]=None, parameters: t.Optional[t.List[CustomAuthParameter]]=None) -> t.Dict:
2
    return self.client.http.post(url=str(self.endpoint / 'proxy'), json={'connectedAccountId': connection_id, 'body': body, 'method': method.upper(), 'endpoint': endpoint, 'parameters': [{'in': param['in_'], 'name': param['name'], 'value': param['value']} for param in parameters or []]}).json()

1
def search_for_a_task(self, use_case: str, limit: t.Optional[int]=None, min_actions_per_task: t.Optional[int]=None, max_actions_per_task: t.Optional[int]=None, apps: t.Optional[t.List[str]]=None) -> t.List[SearchResultTask]:
2
    params: t.Dict[str, t.Any] = {'useCase': use_case}
3
    if limit is not None:
4
        params['limit'] = limit
5
    if min_actions_per_task is not None:
6
        params['minActionsPerTask'] = min_actions_per_task
7
    if max_actions_per_task is not None:
8
        params['maxActionsPerTask'] = max_actions_per_task
9
    if apps is not None:
10
        params['apps'] = ','.join(apps)
11
    response = self._raise_if_required(response=self.client.http.get(str(self.endpoint / 'search' / 'advanced'), params=params))
12
    return [SearchResultTask.model_validate(task) for task in response.json().get('items', [])]

1
def create_file_upload(self, app: str, action: str, filename: str, mimetype: str, md5: str) -> CreateUploadURLResponse:
2
    return CreateUploadURLResponse(**self._raise_if_required(response=self.client.http.post(url=str(self.endpoint / 'files' / 'upload' / 'request'), json={'md5': md5, 'app': app, 'action': action, 'filename': filename, 'mimetype': mimetype})).json())

1
def create(self, app_id: str, name: t.Optional[str]=None, auth_mode: t.Optional['AuthSchemeType']=None, auth_config: t.Optional[t.Dict[str, t.Any]]=None, use_composio_auth: bool=False, force_new_integration: bool=False) -> IntegrationModel:
2
    """
3
        Create a new integration
4
5
        :param app_id: App ID string.
6
        :param name: Name of the integration.
7
        :param auth_param: Auth mode string.
8
        :param auth_config: Authentication configuration.
9
        :param use_composio_auth: Whether to use default composio auth or not
10
        :return: Integration model created by the request.
11
        """
12
    request = {'appId': app_id, 'useComposioAuth': use_composio_auth}
13
    if name is not None:
14
        request['name'] = name
15
    if auth_mode is not None:
16
        request['authScheme'] = auth_mode
17
    if auth_config is not None:
18
        request['authConfig'] = auth_config or {}
19
    if force_new_integration:
20
        request['forceNewIntegration'] = force_new_integration
21
    response = self._raise_if_required(response=self.client.http.post(url=str(self.endpoint), json=request))
22
    return IntegrationModel(**response.json())

1
def remove(self, id: str) -> None:
2
    self.client.http.delete(url=str(self.endpoint / id))

1
@t.overload
2
def get(self, *, page_size: t.Optional[int]=None, page: t.Optional[int]=None, app_id: t.Optional[str]=None, app_name: t.Optional[str]=None, show_disabled: t.Optional[bool]=None) -> t.List[IntegrationModel]:
3
    ...

1
@t.overload
2
def get(self, id: t.Optional[str]=None) -> IntegrationModel:
3
    ...

1
def get(self, id: t.Optional[str]=None, *, page_size: t.Optional[int]=None, page: t.Optional[int]=None, app_id: t.Optional[str]=None, app_name: t.Optional[str]=None, show_disabled: t.Optional[bool]=None) -> t.Union[t.List[IntegrationModel], IntegrationModel]:
2
    if id is not None:
3
        return IntegrationModel(**self._raise_if_required(self.client.http.get(url=str(self.endpoint / id))).json())
4
    quries = {}
5
    if page_size is not None:
6
        quries['pageSize'] = json.dumps(page_size)
7
    if page is not None:
8
        quries['page'] = json.dumps(page)
9
    if app_id is not None:
10
        quries['appId'] = app_id
11
    if app_name is not None:
12
        quries['appName'] = app_name
13
    if show_disabled is not None:
14
        quries['showDisabled'] = json.dumps(show_disabled)
15
    return super().get(queries=quries)

1
@te.deprecated('`get_id` is deprecated, use `get(id=id)`')
2
def get_by_id(self, integration_id: str) -> IntegrationModel:
3
    """
4
        Get an integration by its ID.
5
6
        :param integration_id: Integration ID string.
7
        :return: Integration model.
8
        """
9
    response = self._raise_if_required(self.client.http.get(url=str(self.endpoint / integration_id)))
10
    return IntegrationModel(**response.json())

1
def push(self, record: t.Dict) -> None:
2
    """Push logs to composio."""
3
    if self.client._api_key is None:
4
        return
5
    self.client.http.post(url=str(self.endpoint), json=record)