使用Redis构建简易社交网站(1)-创建用户与动态界面

目的

本文目的：实现简易社交网站中创建新用户和创建新动态功能。（完整代码附在文章末尾）

相关知识

本文将教会你掌握：1.redis基本命令，2.python基本命令。

redis基本命令

hget：从哈希中获取指定域的值。

conn = redis.Redis()
conn.hget("testhash", "field1")

执行结果：2

incr：将 key 中储存的数字值增一。

conn = redis.Redis()
conn.incr("testcount")

执行前：不存在 testcount 键。

执行后：

testcount 1

hset：将哈希中域 field 的值设为 value。

conn = redis.Redis()
conn.hset("testhash", "field1", 2)
conn.hset("testhash", "field2", 4)

执行前：

{'field1': '1'}

执行后：

{'field1': '2', 'field2': '4'}

hmset：同时将多个域-值对设置到哈希表中。

conn = redis.Redis()
conn.hmset("test_hash", {
'id': 1,
'name': 'educoder',
'age': 2,
})

执行后：

{'age': '2', 'id': '1', 'name': 'educoder'}

hincrby：为哈希中指定域的值增加增量 increment，用于统计。

conn = redis.Redis()
conn.hincrby("testhash", "field1", 1)

执行前：

{'field1': '1'}

执行后：

{'field1': '2'}

pipeline：将多条命令按照先后顺序放进一个队列中，一般配合execute一同使用，原子性(atomic)地执行队列里的命令。

conn = redis.Redis()
pipe = conn.pipeline(True) # 事务开始
pipe.incr("counter")
pipe.incr("counter")
pipe.incr("counter")
pipe.execute() # 事务执行

执行结果：[1, 2, 3]，通过下标即可获取对应命令的执行结果。

python基本命令

将字符串全小写化：

'Hello, Educoder'.lower()

执行结果：'hello, educoder'

返回当前时间的时间戳。

time.time()

使用格式化拼接字符串：

"My name is %s, I'm %i years old"%('educoder', 2)

执行结果："My name is educoder, I'm 2 years old"

实战例题：

编写 create_user(login_name, real_name) 函数，实现创建新用户的功能，具体参数与要求如下：

• 方法参数login_name为用户登录名，real_name为用户真名；
• 用户登录名预处理的实现：将用户登录名转换成全小写格式；
• 重名检测的实现：查询哈希键users中是否存在与用户登录名同名的域，若存在，则不允许重新创建该用户，返回None；
• 分配用户编号的实现：对计数器user:id递增1，并将递增后的值作为新用户的编号；
• 存储用户信息的实现：使用事务一次性提交：
  • 存储登录名的实现：将用户登录名记录到哈希键users当中，值为该用户编号；
  • 存储详情的实现：按照如下示意将用户信息存储到哈希键user:{id}中：

• 返回创建结果的实现：返回新创建用户的编号。

编写 create_post(uid, content) 函数，实现创建新动态的功能，具体参数与要求如下：

• 方法参数uid为发布动态的用户编号，content为要发布的动态内容；
• 用户合法性检测的实现：查找用户编号对应详情信息哈希user:{uid}是否存在login_name域，若存在，则记录，若不存在，则不允许创建新动态，返回None；
• 分配动态编号的实现：对计数器post:id递增1，并将递增后的值作为新动态的编号；
• 存储动态信息的实现：按照如下示意将动态信息存储到哈希键post:{id}中：

• 更新用户动态数的实现：为该用户编号对应的详情信息哈希user:{uid}中的posts域的值加1；
• 返回创建结果的实现：返回新创建动态的编号。

 code.py

#code.py
#-*- coding:utf-8 -*-

import re
import time
import redis

conn = redis.Redis()

# 创建新用户
def create_user(login_name, real_name):
    # 请在下面完成要求的功能
    #********* Begin *********#
    login_name = login_name.lower()
    if conn.hget("users", login_name):
        return None

    uid = conn.incr("user:id")
    pipe = conn.pipeline(True)
    pipe.hset("users", login_name, uid)
    pipe.hmset("user:%i"%(uid), {
        'login_name': login_name,
        'id': uid,
        'real_name': real_name,
        'followers': 0,
        'following': 0,
        'posts': 0,
        'last_signup': time.time(),
    })
    pipe.execute()

    return uid
    #********* End *********#

# 为用户创建新动态
def create_post(uid, content):
    # 请在下面完成要求的功能
    #********* Begin *********#
    pipe = conn.pipeline(True)
    pipe.hget("user:%i"%(uid), 'login_name')
    pipe.incr("post:id")
    login_name, pid = pipe.execute()

    if not login_name:
        return None

    pipe.hmset("post:%i"%(pid), {
        'id': pid,
        'uid': uid,
        'content': content,
        'posted': time.time(),
        'user_name': login_name,
    })
    pipe.hincrby("user:%i"%(uid), 'posts')
    pipe.execute()

    return pid
    #********* End *********#

read.py

#read.py
#-*- coding:utf-8 -*-

import os
import sys
import time
import redis
import pprint
from code import *

conn = redis.Redis()
retry_time = 0
while True:
    try:
        conn.ping()
        break
    except redis.exceptions.ConnectionError:
        os.system("redis-server > /dev/null 2>&1 &")
        retry_time += 1
        if retry_time > 3:
            break

pipe = conn.pipeline(True)
pipe.delete("users", "user:id", "post:id")
keys = conn.keys("user:*") + conn.keys("post:*")
for key in keys:
    pipe.delete(key)
pipe.execute()

print "测试 create_user 方法..."
print "第一次创建登录名为 TestUser 的用户"
result1 = create_user('TestUser', 'Test User')
print "创建的用户ID为: " + str(result1)
print "当前分配的用户ID为: " + conn.get("user:id")
user_info = conn.hgetall("user:%i"%(result1))
user_info.pop("last_signup", "404")
print "创建的用户信息为: " + str(user_info)
print

print "第二次创建登录名为 TestUser 的用户"
result2 = create_user('TestUser', 'Test User2')
print "创建的用户ID为: " + str(result2)
print "当前分配的用户ID为: " + conn.get("user:id")
print

print "测试 create_post 方法..."
print "为用户 1 创建一条动态"
pid = create_post(1, "First POST!")
print "创建的动态ID为: " + str(pid)
print "当前分配的动态ID为: " + conn.get("post:id")
post_info = conn.hgetall("post:%i"%(pid))
post_info.pop("posted", "404")
print "创建的动态信息为: " + str(post_info)
user_info = conn.hgetall("user:1")
user_info.pop("last_signup", "404")
print "对应用户信息中更新为: " + str(user_info)
print

print "为不存在的用户 9 创建一条动态"
pid2 = create_post(9, "Illegal POST!")
print "创建的动态ID为: " + str(pid2)

pipe.delete("users", "user:id", "post:id")
keys = conn.keys("user:*") + conn.keys("post:*")
for key in keys:
    pipe.delete(key)
pipe.execute()

测试输入：无；

预期输出：

测试 create_user 方法...
第一次创建登录名为 TestUser 的用户
创建的用户ID为: 1
当前分配的用户ID为: 1
创建的用户信息为: {'login_name': 'testuser', 'posts': '0', 'real_name': 'Test User', 'followers': '0', 'following': '0', 'id': '1'}

第二次创建登录名为 TestUser 的用户
创建的用户ID为: None
当前分配的用户ID为: 1

测试 create_post 方法...
为用户 1 创建一条动态
创建的动态ID为: 1
当前分配的动态ID为: 1
创建的动态信息为: {'content': 'First POST!', 'uid': '1', 'user_name': 'testuser', 'id': '1'}
对应用户信息中更新为: {'login_name': 'testuser', 'posts': '1', 'real_name': 'Test User', 'followers': '0', 'following': '0', 'id': '1'}

为不存在的用户 9 创建一条动态
创建的动态ID为: None

代码007普通

打赏 收藏 海报 链接