Skip to content


Subversion checkout URL

You can clone with
Download ZIP
Fetching contributors…
Cannot retrieve contributors at this time
318 lines (265 sloc) 11.9 KB
#!/usr/bin/env python
# Copyright 2008 FriendFeed
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Methods to interact with the FriendFeed API
Detailed documentation is available at
Many parts of the FriendFeed API require authentication. To support
authentication, FriendFeed gives users a "remote key" that they give to
third party applications to access FriendFeed. The user's nickname and that
remote key are passed as arguments to the constructor of the FriendFeed class,
and the credentials are automatically passed to all called methods. For
session = friendfeed.FriendFeed(nickname, remote_key)
entry = session.publish_message("Testing the FriendFeed API")
Users can get their remote key from You
should direct users who don't know their remote key to that page.
For guidelines on user interface and terminology, check out
import base64
import datetime
import time
import urllib
import urllib2
from google.appengine.api import urlfetch
# We require a JSON parsing library. These seem to be the most popular.
import cjson
parse_json = lambda s: cjson.decode(s.decode("utf-8"), True)
except ImportError:
import simplejson
parse_json = lambda s: simplejson.loads(s.decode("utf-8"))
except ImportError:
import json
parse_json = lambda s: _unicodify(
class FriendFeed(object):
def __init__(self, auth_nickname=None, auth_key=None):
"""Creates a new FriendFeed session for the given user.
The credentials are optional for some operations, but required for
private feeds and all operations that write data, like publish_link.
self.auth_nickname = auth_nickname
self.auth_key = auth_key
def fetch_public_feed(self, **kwargs):
"""Returns the public feed with everyone's public entries.
Authentication is not required.
return self._fetch_feed("/api/feed/public", **kwargs)
def fetch_user_feed(self, nickname, **kwargs):
"""Returns the entries shared by the user with the given nickname.
Authentication is required if the user's feed is not public.
return self._fetch_feed(
"/api/feed/user/" + urllib.quote_plus(nickname), **kwargs)
def fetch_user_comments_feed(self, nickname, **kwargs):
"""Returns the entries the given user has commented on."""
return self._fetch_feed(
"/api/feed/user/" + urllib.quote_plus(nickname) + "/comments",
def fetch_user_likes_feed(self, nickname, **kwargs):
"""Returns the entries the given user has "liked"."""
return self._fetch_feed(
"/api/feed/user/" + urllib.quote_plus(nickname) + "/likes",
def fetch_user_discussion_feed(self, nickname, **kwargs):
"""Returns the entries the given user has commented on or "liked"."""
return self._fetch_feed(
"/api/feed/user/" + urllib.quote_plus(nickname) + "/discussion",
def fetch_multi_user_feed(self, nicknames, **kwargs):
"""Returns a merged feed with all of the given users' entries.
Authentication is required if any one of the users' feeds is not
return self._fetch_feed("/api/feed/user", nickname=",".join(nicknames),
def fetch_home_feed(self, **kwargs):
"""Returns the entries the authenticated user sees on their home page.
Authentication is always required.
return self._fetch_feed("/api/feed/home", **kwargs)
def search(self, q, **kwargs):
"""Searches over entries in FriendFeed.
If the request is authenticated, the default scope is over all of the
entries in the authenticated user's Friends Feed. If the request is
not authenticated, the default scope is over all public entries.
The query syntax is the same syntax as
kwargs["q"] = q
return self._fetch_feed("/api/feed/search", **kwargs)
def publish_message(self, message, **kwargs):
"""Publishes the given message to the authenticated user's feed.
See publish_link for additional options.
return self.publish_link(title=message, link=None, **kwargs)
def publish_link(self, title, link, comment=None, image_urls=[],
images=[], via=None):
"""Publishes the given link/title to the authenticated user's feed.
Authentication is always required.
image_urls is a list of URLs that will be downloaded and included as
thumbnails beneath the link. The thumbnails will all link to the
destination link. If you would prefer that the images link somewhere
else, you can specify images[] instead, which should be a list of
dicts of the form {"url": ..., "link": ...}. The thumbnail with the
given url will link to the specified link.
We return the parsed/published entry as returned from the server, which
includes the final thumbnail URLs as well as the ID for the new entry.
session = friendfeed.FriendFeed(nickname, remote_key)
entry = session.publish_link(
title="Testing the FriendFeed API",
print "Posted images at" % entry["id"]
post_args = {"title": title}
if link:
post_args["link"] = link
if comment:
post_args["comment"] = comment
if via:
post_args["via"] = via
for image_url in image_urls:
images.append({"url": image_url})
for i, image in enumerate(images):
post_args["image%d_url" % i] = image["url"]
if image.get("link"):
post_args["image%d_link" % i] = image["link"]
feed = self._fetch_feed("/api/share", post_args=post_args)
return feed["entries"][0]
def add_comment(self, entry_id, body, via=None):
"""Adds the given comment to the entry with the given ID.
We return the ID of the new comment, which can be used to edit or
delete the comment.
args = {
"entry": entry_id,
"body": body
if via: args["via"] = via
result = self._fetch("/api/comment", args)
return result["id"]
def edit_comment(self, entry_id, comment_id, body):
"""Updates the comment with the given ID."""
self._fetch("/api/comment", {
"entry": entry_id,
"comment": comment_id,
"body": body
def delete_comment(self, entry_id, comment_id):
"""Deletes the comment with the given ID."""
self._fetch("/api/comment/delete", {
"entry": entry_id,
"comment": comment_id,
def undelete_comment(self, entry_id, comment_id):
"""Un-deletes the comment with the given ID."""
self._fetch("/api/comment/delete", {
"entry": entry_id,
"comment": comment_id,
"undelete": 1,
def add_like(self, entry_id):
"""'Likes' the entry with the given ID."""
self._fetch("/api/like", {
"entry": entry_id,
def delete_like(self, entry_id):
"""Deletes the 'Like' for the entry with the given ID (if any)."""
self._fetch("/api/like/delete", {
"entry": entry_id,
def _fetch_feed(self, uri, post_args=None, **kwargs):
"""Publishes to the given URI and parses the returned JSON feed."""
# Parse all the dates in the result JSON
result = self._fetch(uri, post_args, **kwargs)
rfc3339_date = "%Y-%m-%dT%H:%M:%SZ"
date_properties = frozenset(("updated", "published"))
for entry in result.get("entries", []):
entry["updated"] = self._parse_date(entry["updated"])
entry["published"] = self._parse_date(entry["published"])
for comment in entry.get("comments", []):
comment["date"] = self._parse_date(comment["date"])
for like in entry.get("likes", []):
like["date"] = self._parse_date(like["date"])
return result
def _fetch(self, uri, post_args, **url_args):
url_args["format"] = "json"
args = urllib.urlencode(url_args)
url = "" + uri + "?" + args
if post_args is not None:
request = urlfetch.fetch(url, urllib.urlencode(post_args))
request = urlfetch.fetch(url)
if self.auth_nickname and self.auth_key:
pair = "%s:%s" % (self.auth_nickname, self.auth_key)
token = base64.b64encode(pair)
request.add_header("Authorization", "Basic %s" % token)
return parse_json(request.content)
def _parse_date(self, date_str):
rfc3339_date = "%Y-%m-%dT%H:%M:%SZ"
return datetime.datetime(*time.strptime(date_str, rfc3339_date)[:6])
def _unicodify(json):
"""Makes all strings in the given JSON-like structure unicode."""
if isinstance(json, str):
return json.decode("utf-8")
elif isinstance(json, dict):
for name in json:
json[name] = _unicodify(json[name])
elif isinstance(json, list):
for part in json:
return json
def _example():
# Fill in a nickname and a valid remote key below for authenticated
# actions like posting an entry and reading a protected feed
# session = FriendFeed(auth_nickname=nickname, auth_key=remote_key)
session = FriendFeed()
feed = session.fetch_public_feed()
# feed = session.fetch_user_feed("bret")
# feed = session.fetch_user_feed("paul", service="twitter")
# feed = session.fetch_user_discussion_feed("bret")
# feed = session.fetch_multi_user_feed(["bret", "paul", "jim"])
# feed ="who:bret friendfeed")
for entry in feed["entries"]:
print entry["published"].strftime("%m/%d/%Y"), entry["title"]
if session.auth_nickname and session.auth_key:
# The feed that the authenticated user would see on their home page
feed = session.fetch_home_feed()
# Post a message on this user's feed
entry = session.publish_message("Testing the FriendFeed API")
print "Posted new message at" % entry["id"]
# Post a link on this user's feed
entry = session.publish_link(title="Testing the FriendFeed API",
print "Posted new link at" % entry["id"]
# Post a link with two thumbnails on this user's feed
entry = session.publish_link(
title="Testing the FriendFeed API",
print "Posted images at" % entry["id"]
if __name__ == "__main__":
Jump to Line
Something went wrong with that request. Please try again.