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.
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.
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:
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
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.
I saw this fun tool on YC News called pysnooper: https://news.ycombinator.com/item?id=19717786 . I’ve been trying it out all day on various Python applications, and it’s actually making debugging fun and a whole lot easier!
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):
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))
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
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"