a stupid twitter implementation using redis + Python

.gitignore

*.pyc

Twittis.py

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())

TwittisTest.py

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()