Skip to content

Instantly share code, notes, and snippets.

@jpcaruana
Created November 24, 2011 15:49
Show Gist options
  • Save jpcaruana/1391648 to your computer and use it in GitHub Desktop.
Save jpcaruana/1391648 to your computer and use it in GitHub Desktop.
a stupid twitter implementation using redis + Python
import time
import redis
class Utilisateur:
def __init__(self, prenom, nom, id=None):
self.prenom = prenom
self.nom = nom
if id:
self.id = int(id)
else:
self.id = repository.cree_utilisateur(prenom, nom)
def sAbonne(self, followee):
repository.sAbonne(self, followee)
def suiveurs(self):
return repository.suiveurs(self)
def amis(self):
return repository.amis(self)
def en_commun(self, utilisateur):
return repository.en_commun(self, utilisateur)
def dit(self, tweet):
repository.dit(self, tweet)
def tweets(self):
return repository.tweets(self)
def __str__(self):
return '%s (%s) %s' % (self.prenom, self.id, self.nom)
def __eq__(self, autre):
return self.id == autre.id
class Repository:
def __init__(self):
self.redis = connection()
def cree_utilisateur(self, prenom, nom):
utilisateur_id = self.redis.incr('next:user:id')
clef = 'user:%d' % (utilisateur_id)
self.redis.hmset(clef, {'prenom': prenom, 'nom': nom})
return utilisateur_id
def recherche_utilisateur(self, id):
utilisateur = self.redis.hgetall('user:%s' % id)
return Utilisateur(utilisateur['prenom'], utilisateur['nom'], id)
def recherche_utilisateurs(self, utilisateur_ids):
return [ self.recherche_utilisateur(id) for id in utilisateur_ids ]
def sAbonne(self, follower, followee):
self.redis.sadd('user:%d:sAbonne' % follower.id, followee.id)
self.redis.sadd('user:%d:suiveurs' % followee.id, follower.id)
def suiveurs(self, utilisateur):
return self.recherche_utilisateurs(self.redis.smembers('user:%d:suiveurs' % utilisateur.id))
def amis(self, utilisateur):
return self.recherche_utilisateurs(self.redis.sinter('user:%d:suiveurs' % utilisateur.id, 'user:%d:sAbonne' % utilisateur.id))
def en_commun(self, utilisateur, other):
return self.recherche_utilisateurs(self.redis.sinter('user:%d:sAbonne' % utilisateur.id, 'user:%d:sAbonne' % other.id))
def dit(self, utilisateur, tweet):
tweet_id = self.redis.incr('next:tweet:id')
followers = utilisateur.suiveurs()
with self.redis.pipeline() as pipe:
pipe.multi()
pipe.set('tweet:%d' % tweet_id, '%s|%s|%s' % (utilisateur.id, time.time(), tweet))
pipe.lpush('user:%d:tweets' % utilisateur.id, tweet_id)
for follower in followers:
pipe.lpush('user:%s:tweets' % follower.id, tweet_id)
pipe.lpush('global:timeline', tweet_id)
pipe.ltrim('global:timeline', 0, 1000)
pipe.execute()
def tweets(self, utilisateur):
tweet_ids = self.redis.lrange('user:%d:tweets' % utilisateur.id, 0, -1)
return [ self.redis.get('tweet:%s' % id_tweet) for id_tweet in tweet_ids ]
def connection():
return redis.Redis(connection_pool=pool)
pool = redis.ConnectionPool(host='localhost', port=6379, db=1)
repository = Repository()
def affiche_une_liste(elements):
return ', '.join(str(element) for element in elements)
if __name__ == '__main__':
connection().flushdb()
jp = Utilisateur('JP', 'Caruana')
bruno = Utilisateur('Bruno', 'Thomas')
jb = Utilisateur('Jean-Baptiste', 'Potonnier')
bruno.sAbonne(jp)
jb.sAbonne(jp)
jp.sAbonne(jb)
print "%s est suivi par %s" % (jp, affiche_une_liste(jp.suiveurs()))
print "%s est ami avec %s car ils se suivent mutuellement" % (jp, affiche_une_liste(jp.amis()))
print "%s et %s suivent tous les 2 %s" % (jb, bruno, affiche_une_liste(jb.en_commun(bruno)))
for i in range(5):
jp.dit("%d Coucou, je m'appelle %s" % (i, str(jp)))
bruno.dit("Ici Bruno, bien content d'etre ici")
print "%s voit les tweets suivants : %s" % (bruno, bruno.tweets())
import unittest
from Twittis import *
class TwittisTest(unittest.TestCase):
def setUp(self):
self.redis = connection()
self.redis.flushdb()
self.jp = Utilisateur('JP', 'Caruana')
self.bruno = Utilisateur('Bruno', 'Thomas')
self.bruno.sAbonne(self.jp)
def tearDown(self):
self.redis.flushdb()
def test_affiche_suiveurs(self):
self.assertEqual(self.bruno, self.jp.suiveurs()[0])
def test_amis_se_suivent_mutuellement(self):
self.jp.sAbonne(self.bruno)
self.assertEqual(self.bruno, self.jp.amis()[0])
def test_un_tweet_est_vu_par_son_auteur_et_ses_suiveurs(self):
self.jp.dit("tweet")
self.assertTrue('tweet' in self.jp.tweets()[0])
self.assertTrue('tweet' in self.bruno.tweets()[0])
def test_amis_en_commun(self):
self.jb = Utilisateur('Jean-Baptiste', 'Potonnier')
self.jb.sAbonne(self.jp)
self.assertEqual(self.jp, self.jb.en_commun(self.bruno)[0])
if __name__ == "__main__":
unittest.main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment