Tweets

This is a module for interacting with Twitter tweets

get_status(status_id, twitter_credentials, **kwargs) async

Returns a single status specified by the ID parameter.

Parameters:

Name	Type	Description	Default
status_id	int	The ID of the status.	required
twitter_credentials	TwitterCredentials	Credentials to use for authentication with Twitter.	required
kwargs	dict	Additional keyword arguments to pass to get_status.	{}

Example

Tweets an update with just text.

from prefect import flow
from prefect_twitter import TwitterCredentials
from prefect_twitter.tweets import get_status

@flow
def example_get_status_flow():
    twitter_credentials = TwitterCredentials(
        consumer_key=consumer_key,
        consumer_secret=consumer_secret,
        access_token=access_token,
        access_token_secret=access_token_secret
    )
    status_id = 1504591031626571777
    status = get_status(status_id, twitter_credentials)
    return status

example_get_status_flow()

Source code in prefect_twitter/tweets.py

@task
async def get_status(
    status_id: int, twitter_credentials: "TwitterCredentials", **kwargs: dict
) -> "Status":
    """
    Returns a single status specified by the ID parameter.

    Args:
        status_id: The ID of the status.
        twitter_credentials: Credentials to use for authentication with Twitter.
        kwargs: Additional keyword arguments to pass to
            [get_status](https://docs.tweepy.org/en/stable/api.html#tweepy.API.get_status).
    Returns:
        The Status object.

    Example:
        Tweets an update with just text.
        ```python
        from prefect import flow
        from prefect_twitter import TwitterCredentials
        from prefect_twitter.tweets import get_status

        @flow
        def example_get_status_flow():
            twitter_credentials = TwitterCredentials(
                consumer_key=consumer_key,
                consumer_secret=consumer_secret,
                access_token=access_token,
                access_token_secret=access_token_secret
            )
            status_id = 1504591031626571777
            status = get_status(status_id, twitter_credentials)
            return status

        example_get_status_flow()
        ```
    """
    api = twitter_credentials.get_api()
    partial_get = partial(api.get_status, status_id, **kwargs)
    status = await to_thread.run_sync(partial_get)
    return status

update_status(twitter_credentials, status=None, media_ids=None, **kwargs) async

Updates the authenticating user's current status, also known as Tweeting.

Parameters:

Name	Type	Description	Default
twitter_credentials	TwitterCredentials	Credentials to use for authentication with Twitter.	required
status	Optional[str]	Text of the Tweet being created. This field is required if media_ids is not present.	None
media_ids	Optional[List[Union[int, str]]]	A list of Media IDs being attached to the Tweet.	None
kwargs	dict	Additional keyword arguments to pass to update_status.	{}

Example

Tweets an update with just text.

from prefect import flow
from prefect_twitter import TwitterCredentials
from prefect_twitter.tweets import update_status

@flow
def example_update_status_flow():
    twitter_credentials = TwitterCredentials(
        consumer_key="consumer_key",
        consumer_secret="consumer_secret",
        access_token="access_token",
        access_token_secret="access_token_secret"
    )
    status_id = update_status(twitter_credentials, status="A Prefect Tweet!")
    return status_id

example_update_status_flow()

Tweets an update with text and a media.

from prefect import flow
from prefect_twitter import TwitterCredentials
from prefect_twitter.tweets import update_status
from prefect_twitter.media import media_upload

@flow
def example_update_status_flow():
    twitter_credentials = TwitterCredentials(
        consumer_key="consumer_key",
        consumer_secret="consumer_secret",
        access_token="access_token",
        access_token_secret="access_token_secret"
    )
    media_id = media_upload("prefect.png", twitter_credentials)
    status_id = update_status(
        twitter_credentials,
        status="Prefect!",
        media_ids=[media_id]
    )
    return status_id

example_update_status_flow()

Source code in prefect_twitter/tweets.py

@task
async def update_status(
    twitter_credentials: "TwitterCredentials",
    status: Optional[str] = None,
    media_ids: Optional[List[Union[int, str]]] = None,
    **kwargs: dict
) -> int:
    """
    Updates the authenticating user's current status, also known as Tweeting.

    Args:
        twitter_credentials: Credentials to use for authentication with Twitter.
        status: Text of the Tweet being created. This field is required
            if media_ids is not present.
        media_ids: A list of Media IDs being attached to the Tweet.
        kwargs: Additional keyword arguments to pass to
            [update_status](https://docs.tweepy.org/en/stable/api.html#tweepy.API.update_status).
    Returns:
        The status ID.

    Example:
        Tweets an update with just text.
        ```python
        from prefect import flow
        from prefect_twitter import TwitterCredentials
        from prefect_twitter.tweets import update_status

        @flow
        def example_update_status_flow():
            twitter_credentials = TwitterCredentials(
                consumer_key="consumer_key",
                consumer_secret="consumer_secret",
                access_token="access_token",
                access_token_secret="access_token_secret"
            )
            status_id = update_status(twitter_credentials, status="A Prefect Tweet!")
            return status_id

        example_update_status_flow()
        ```

        Tweets an update with text and a media.
        ```python
        from prefect import flow
        from prefect_twitter import TwitterCredentials
        from prefect_twitter.tweets import update_status
        from prefect_twitter.media import media_upload

        @flow
        def example_update_status_flow():
            twitter_credentials = TwitterCredentials(
                consumer_key="consumer_key",
                consumer_secret="consumer_secret",
                access_token="access_token",
                access_token_secret="access_token_secret"
            )
            media_id = media_upload("prefect.png", twitter_credentials)
            status_id = update_status(
                twitter_credentials,
                status="Prefect!",
                media_ids=[media_id]
            )
            return status_id

        example_update_status_flow()
        ```
    """  # noqa
    num_media = len(media_ids) if media_ids else 0
    logger = get_run_logger()
    logger.info("Updating status with %s media.", num_media)

    if not status and not media_ids:  # `not` checks for None and []
        raise ValueError("One of text or media_ids must be provided")

    api = twitter_credentials.get_api()
    partial_update = partial(
        api.update_status, status=status, media_ids=media_ids, **kwargs
    )
    status = await to_thread.run_sync(partial_update)
    return status.id