Skip to content

Media

This is a module for interacting with Twitter media

get_media_upload_status(media_id, twitter_credentials) async

Check on the progress of a chunked media upload. If the upload has succeeded, it's safe to create a Tweet with this media_id.

Parameters:

Name Type Description Default
media_id int

The ID of the media to check.

 required
twitter_credentials TwitterCredentials

Credentials to use for authentication with Twitter.

 required

Returns:

Type Description
Media

The Media object.
Example

Tweets an update with just text. 

from prefect import flow
from prefect_twitter import TwitterCredentials
from prefect_twitter.media import get_media_upload_status

@flow
def example_get_media_upload_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 = 1443668738906234883
    media = get_media_upload_status(media_id, twitter_credentials)
    return media

example_get_media_upload_status_flow()

Source code in prefect_twitter/media.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@task
async def get_media_upload_status(
    media_id: int, twitter_credentials: "TwitterCredentials"
) -> "Media":
    """
    Check on the progress of a chunked media upload. If the upload has succeeded,
    it's safe to create a Tweet with this media_id.

    Args:
        media_id: The ID of the media to check.
        twitter_credentials: Credentials to use for authentication with Twitter.

    Returns:
        The Media object.

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

        @flow
        def example_get_media_upload_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 = 1443668738906234883
            media = get_media_upload_status(media_id, twitter_credentials)
            return media

        example_get_media_upload_status_flow()
        ```
    """
    api = twitter_credentials.get_api()
    partial_get = partial(api.get_media_upload_status, media_id)
    media = await to_thread.run_sync(partial_get)
    return media

media_upload(filename, twitter_credentials, file=None, chunked=False, **kwargs) async

Uploads media to Twitter. Chunked media upload is automatically used for videos.

Parameters:

Name Type Description Default
filename Union[Path, str]

The filename of the image to upload. This field is used for MIME type detection.

 required
twitter_credentials TwitterCredentials

Credentials to use for authentication with Twitter.

 required
file Optional[IOBase]

A file object to upload. If not specified, this task will attempt to locate and upload a file with the name specified in filename.

 None
chunked bool

Whether or not to use chunked media upload. Videos use chunked upload regardless of this parameter.

 False
kwargs dict

Additional keyword arguments to pass to media_upload.

 {}
Example

Uploads an image from a file path to Twitter. 

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

@flow
def example_media_upload_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("/path/to/prefection.jpg", twitter_credentials)
    return media_id

example_media_upload_flow()

Source code in prefect_twitter/media.py 
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
@task
async def media_upload(
    filename: Union[Path, str],
    twitter_credentials: "TwitterCredentials",
    file: Optional["IOBase"] = None,
    chunked: bool = False,
    **kwargs: dict
) -> int:
    """
    Uploads media to Twitter. Chunked media upload
    is automatically used for videos.

    Args:
        filename: The filename of the image to upload.
            This field is used for MIME type detection.
        twitter_credentials: Credentials to use for authentication with Twitter.
        file: A file object to upload. If not specified, this task will attempt to
            locate and upload a file with the name specified in filename.
        chunked: Whether or not to use chunked media upload.
            Videos use chunked upload regardless of this parameter.
        kwargs: Additional keyword arguments to pass to
            [media_upload](https://docs.tweepy.org/en/stable/api.html#tweepy.API.media_upload).
    Returns:
        The media ID.

    Example:
        Uploads an image from a file path to Twitter.
        ```python
        from prefect import flow
        from prefect_twitter import TwitterCredentials
        from prefect_twitter.media import media_upload

        @flow
        def example_media_upload_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("/path/to/prefection.jpg", twitter_credentials)
            return media_id

        example_media_upload_flow()
        ```
    """  # noqa
    logger = get_run_logger()
    logger.info("Uploading media named %s", filename)

    api = twitter_credentials.get_api()
    partial_media = partial(
        api.media_upload, filename=filename, file=file, chunked=chunked, **kwargs
    )
    media = await to_thread.run_sync(partial_media)
    media_id = media.media_id
    return media_id