Python: Changing Timezones In A DateTime

Here’s some quick code samples for shifting a UTC datetime object (created_at is a datetime.utcnow()) to a different timezone. In this first example, we use timedelta to add/remove hours to find the current time at UTC-6:00.

local_time = created_at + timedelta(hours=-6)
final_time =  datetime.strftime(local_time, '%Y-%m-%d %I:%M:%S %p')

In this sample, a datetime (created_at) is declared to be a UTC time, then converted into US/Chicago time and formatted for human presentation:

local_datetime = pytz.timezone('UTC').localize(created_at).astimezone(pytz.timezone('US/Central'))
local_datetime_str = "Created (User Local Time): " +  str(datetime.strftime(local_datetime, '%Y-%m-%d %I:%M:%S %p'))

Correcting A SQLite Code Example

I’ve been experimenting with filtering and manipulating a large amount of data within a Google Cloud Function. I decided to use an in-memory SQLite database to help manage all the data, so I googled up some code samples. This page came up with some helpful Python code samples.

Unfortunately when I tried to run the sample code, Cloud Functions popped an error. The sample code uses Python 2-style print as a statement instead of as a function call – i.e. the print call is missing the parentheses needed to make it a correct function call. Here’s a sample screenshot:

I’ve placed red arrows next to the erroneous print statements. If you paste this code into Google Cloud Functions, it won’t work because print needs to be a function call, (with parentheses) instead of a statement (missing parentheses). Credit: https://www.tutorialspoint.com/sqlite/sqlite_python.htm

Below is a fixed version of the code in the linked page. You can paste it directly into the Google Cloud Functions editor and it’ll work: it sets up an in-memory database, creates a table, adds data, then queries data out of it.

import sqlite3



def hello_world(request):
    """Responds to any HTTP request.
    Args:
        request (flask.Request): HTTP request object.
    Returns:
        The response text or any set of values that can be turned into a
        Response object using
        `make_response <http://flask.pocoo.org/docs/1.0/api/#flask.Flask.make_response>`.
    """
    conn = sqlite3.connect(":memory:")
    conn.execute('''CREATE TABLE COMPANY
         (ID INT PRIMARY KEY     NOT NULL,
         NAME           TEXT    NOT NULL,
         AGE            INT     NOT NULL,
         ADDRESS        CHAR(50),
         SALARY         REAL);''')
    conn.execute("INSERT INTO COMPANY (ID,NAME,AGE,ADDRESS,SALARY) \
        VALUES (1, 'Paul', 32, 'California', 20000.00 )");
    conn.execute("INSERT INTO COMPANY (ID,NAME,AGE,ADDRESS,SALARY) \
        VALUES (2, 'Allen', 25, 'Texas', 15000.00 )");
    conn.execute("INSERT INTO COMPANY (ID,NAME,AGE,ADDRESS,SALARY) \
        VALUES (3, 'Teddy', 23, 'Norway', 20000.00 )");
    conn.execute("INSERT INTO COMPANY (ID,NAME,AGE,ADDRESS,SALARY) \
        VALUES (4, 'Mark', 25, 'Rich-Mond ', 65000.00 )");
    conn.commit()
    print("Records created successfully");
    cursor = conn.execute("SELECT id, name, address, salary from COMPANY")
    for row in cursor:
        print("ID = ", row[0])
        print("NAME = ", row[1])
        print("ADDRESS = ", row[2])
        print("SALARY = ", row[3], "\n")
    conn.close()
    request_json = request.get_json()
    if request.args and 'message' in request.args:
        return request.args.get('message')
    elif request_json and 'message' in request_json:
        return request_json['message']
    else:
        return f'Hello World!'

Use this code as a starting point to build your own cloud functions and work with data.

I’m pleasantly surprised at how fast SQLite runs within a cloud function – I was worried that the function would run out of memory quickly, but I’ve been manipulating thousands of rows comfortably within a 512MB RAM function.

NewsBlur: Iterating Through A Folder’s RSS Feed

After Google Reader was shut down, I moved to NewsBlur to follow my RSS feeds. The great thing about NewsBlur is that you can add RSS feeds to a folder and Newsblur will merge all the stories under that folder into a single RSS feed.

Under NewsBlur, you’ll want to pull the folder RSS feed from the settings option:

NewsBlur settings option - the folder RSS URL is at the bottom.

The following Python code can pull the feed and iterate through it to find article information. At the bottom of this code example, each child represents a possible article, and sub_child represents a property on the article: the URL, the title, etc. I use a variant of this code to help identify important news stories.

import requests
import xml.etree.ElementTree as ET
import logging
import datetime, pytz
import json
import urllib.parse

#tears through the newsblur folder xml searching for <entry> items
def parse_newsblur_xml():
    r = requests.get('NEWSBLUR_FOLDER_RSS')
    if r.status_code != 200:
        print("ERROR: Unable to retrieve address ")
        return "error"
    xml = r.text
    xml_root = ET.fromstring(xml)
    #we search for <entry> tags because each entry tag stores a single article from a RSS feed
    for child in xml_root:
        if not child.tag.endswith("entry"):
            continue
        #if we are down here, the tag is an entry tag and we need to parse out info
        #Grind through the children of the <entry> tag
        for sub_child in child:
            if sub_child.tag.endswith("category"): #article categories
                #call sub_child.get('term') to get categories of this article
            elif sub_child.tag.endswith("title"): #article title
                #call sub_child.text to get article title
            elif sub_child.tag.endswith("summary"): #article summary
                #call sub_child.text to get article summary
            elif sub_child.tag.endswith("link"):
                #call sub_child.get('href') to get article URL

IBM Watson Natural Language Understanding

I was fiddling with code that worked with IBM Watson’s Natural Language Understanding API, and kept getting the following error:

{
    "error": "invalid request: content is empty",
    "code": 400
}

What happened is that I was calling out to IBM using the following requests Python code (post_data represents a Python object being encoded to JSON):

post_data_string = json.dumps(post_data)
requests.post(endpoint, data=post_data_string, timeout=45, auth=(ibm_user, ibm_pass))

However, it seems that the API insists on having the correct request content type set; i.e. you must set Content-Type: application/json for the IBM Watson servers to notice there is a data body in the POST request. I fixed it by using the json parameter – the requests library for Python automatically inserts the application/json content type when this parameter is used. If you use a different language, you’ll need to set the proper content type in the language’s preferred manner.

requests.post(endpoint, json=post_data, timeout=45, auth=(ibm_user, ibm_pass))

PhantomJSCloud Error: Invalid Character ‘u’ – Python

I use PhantomJSCloud to take screenshots of web apps. While writing new code, I noticed this error coming back from the server:

{  
   "name":"HttpStatusCodeException",
   "statusCode":400,
   "message":"Error extracting userRequest. innerException: JSON5: invalid character 'u' at 1:1",
   "stack":[  
      "no stack unless env is DEV or TEST, or logLevel is DEBUG or TRACE"
   ]
}

The problem came because the request wasn’t JSON-encoding the object; the fix looks like this (using the requests library):

    post_data_string = json.dumps(python_object_here, sort_keys=True, indent=3, separators=(',', ': '))
    r = requests.post('https://phantomjscloud.com/api/browser/v2/', data=post_data_string)

Tweepy Code Sample: Auth & Iterating Through Following Users

Here’s a short code example using Tweepy to pull a list of following users (users that you follow). consumer_key, consumer_secret, access_token and access_token_secret are necessary tokens for authenticating into Twitter.

auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)

api = tweepy.API(auth)

for friend in tweepy.Cursor(api.friends).items(3):
    # Extract friend (following) Twitter screen name and user id
    friend_screen_name = friend.screen_name
    friend_id = friend.id_str
    print("Friend screen name & ID: %s - %s" % (friend_screen_name, friend_id))

Creating A WordPress Blog Slug Part 2 – Python

In a previous post, I posted a sample NodeJS function to assemble a WordPress blog slug. I ended up rewriting part of the larger application (and the function itself) in Python.

In the below function, source is a blog title string, and it returns a slug suitable for use in a blog URL.

def generate_slug(source):
    i = 0
    source = source.lower().strip()
    allowed = "abcdefghijklmnopqrstuvwxyz1234567890"
    slug = ""
    while i < (len(source) - 0):
        single_letter = source[i:i+1]
        if single_letter in allowed:
            slug += single_letter
        elif not slug.endswith("-"):
            #letter is not allowed
            #check that the slug doesn't already end in a dash
            slug += "-"
        i = i + 1
    return slug

Listing Files Within A Bucket Folder – Python

Here’s a short code example in Python to iterate through a folder’s ( thisisafolder ) contents within Google Cloud Storage (GCS). Each filename can be accessed through blobi.name – in the below code sample, we print it out and test whether it ends with .json.

Remember that folders don’t actually exist on GCS, but a folder-like structure can be created by prefixing filenames with the folder name and the forward slash character ( / ).

    client = storage.Client()
    bucket = client.get_bucket("example-bucket-name")
    blob_iterator = bucket.list_blobs(prefix="thisisafolder",client=client)
    #iterate through and print out blob filenames
    for blobi in blob_iterator:
        print(blobi.name)
        if blobi.name.endswith(".json"):
            #do something with blob that ends with ".json"