Skip to content

Twitter api

post_tweet(tweet_text, pkg_id)

Attempts to post the tweet. Returns a boolean success variable and a message describing the reason for the failure/success in posting the tweet.

Parameters:

Name Type Description Default
tweet_text

The text to post. This is passed in rather than generated inside the method to allow users to change the tweet before posting (if enabled).

 required
pkg_id

The package ID (for caching).

 required

Returns:

Type Description

boolean, str
Source code in ckanext/twitter/lib/twitter_api.py 
 71
 72
 73
 74
 75
 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
def post_tweet(tweet_text, pkg_id):
    """
    Attempts to post the tweet. Returns a boolean success variable and a message
    describing the reason for the failure/success in posting the tweet.

    :param tweet_text: The text to post. This is passed in rather than
    generated inside the method to allow users to change the tweet before
    posting (if enabled).
    :param pkg_id: The package ID (for caching).
    :return: boolean, str
    """
    logger.info('ckanext-twitter has been deprecated; please consider removing it')

    if config_helpers.twitter_is_debug():
        logger.debug(f'Not posted (debug): {tweet_text}')
        return False, 'debug'

    # if not enough time has passed since the last tweet
    if not cache_helpers.expired(pkg_id):
        logger.debug(f'Not posted (insufficient rest period): {tweet_text}')
        return False, 'insufficient rest period'

    # if we can't authenticate
    if not twitter_authenticate():
        logger.debug(f'Not posted (not authenticated): {tweet_text}')
        return False, 'not authenticated'

    # try to actually post
    client = twitter_client()
    url = 'https://api.twitter.com/1.1/statuses/update.json'
    params = {'status': tweet_text}
    request = oauth2.Request(method='POST', url=url, parameters=params)
    postdata = request.to_postdata()
    response, content = client.request(url, 'POST', postdata)
    if response.status == 200:
        cache_helpers.cache(pkg_id)
        logger.debug(f'Posted successfully: {tweet_text}')
    else:
        logger.debug(f'Not posted (tweet unsuccessful): {tweet_text}')
    return response.status == 200, f'{response.status} {response.reason}'

twitter_authenticate()

Verifies that the client is able to connect to the twitter API.

Refreshes any unauthenticated cached client.

Returns:

Type Description

boolean
Source code in ckanext/twitter/lib/twitter_api.py 
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
def twitter_authenticate():
    """
    Verifies that the client is able to connect to the twitter API.

    Refreshes any unauthenticated cached client.
    :return: boolean
    """
    logger.info('ckanext-twitter has been deprecated; please consider removing it')

    authenticated = False
    while not authenticated:
        client = twitter_client()
        url = 'https://api.twitter.com/1.1/account/verify_credentials.json'
        response, content = client.request(url, 'GET')
        authenticated = response.status == 200
        if not authenticated:
            # if the client isn't in the cache we don't care
            with suppress(KeyError):
                cache_helpers.cache_manager.invalidate(twitter_client)
            break
    if authenticated:
        if any(
            [
                c.startswith('no-') and c.endswith('-set')
                for c in config_helpers.twitter_get_credentials()
            ]
        ):
            authenticated = False
    return authenticated

twitter_client()

Attempts to create a client for accessing the twitter API using the credentials specified in the configuration file. Does not test for success. Caches the resulting client in the 'twitter' cache.

Returns:

Type Description

oauth2.Client
Source code in ckanext/twitter/lib/twitter_api.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@cache_helpers.cache_manager.cache('twitter', 'client')
def twitter_client():
    """
    Attempts to create a client for accessing the twitter API using the credentials
    specified in the configuration file. Does not test for success. Caches the resulting
    client in the 'twitter' cache.

    :return: oauth2.Client
    """
    logger.info('ckanext-twitter has been deprecated; please consider removing it')

    (
        consumer_key,
        consumer_secret,
        token_key,
        token_secret,
    ) = config_helpers.twitter_get_credentials()
    consumer = oauth2.Consumer(consumer_key, consumer_secret)
    token = oauth2.Token(token_key, token_secret)
    client = oauth2.Client(consumer, token)
    return client