Skip to content

prefect_twitter.tweets

This is a module for interacting with Twitter tweets

get_status 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

Additional keyword arguments to pass to get_status.

required

Returns:

Type Description
'Status'

The Status object.

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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@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 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

Additional keyword arguments to pass to update_status.

required

Returns:

Type Description
int

The status ID.

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
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@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