第6课：datetime模块操作数据库__name__redismock接口

Posted 2020-10-21 茄子子

1.  datetime模块

import datetime

print(datetime.datetime.today())  # 当前时间 2018-01-23 17:22:35.739667
print(datetime.datetime.now())  # 和today一样 2018-01-23 17:22:35.739667

print(datetime.datetime.today().strftime(\'%H:%M:%S\'))  # 按指定格式格式化好的时间字符串
print(datetime.datetime.today() + datetime.timedelta(-3))  # 取三天前的 三天后的（3） 2018-01-20 17:25:37.470085
print(datetime.date.today())  # 取当天的日期 2018-01-23

2. 操作mysql数据库

import pymysql
import config
# 1. 连接mysql，ip 端口号 用户名 密码 数据库
# 2. 建立游标
# 3. 执行sql
# 4. 获取结果
# 5. 关闭游标
# 6. 关闭连接

# charset必须写utf8，写utf-8会报错
conn = pymysql.connect(host=config.host, port=config.port, user=config.user, passwd=config.passwd,
                       db=config.dbname, charset=\'utf8\')
cur = conn.cursor()  # 建立游标，这种方式获取到的返回结果是元组
# cur = conn.cursor(cursor=pymysql.cursors.DictCursor)  # 这种方式获取到的结果是字典类型的
sql = \'select * from bt_stu limit 2;\'
# sql = "insert into bt_stu values(560, \'qq\', 1, 18101000001, \'中国\', 1);"
cur.execute(sql)  # 执行sql
conn.commit()
res = cur.fetchall()  # 获取sql语句执行的结果，是一个二维的元组
# res1 = cur.fetchone()  # 只获取一条结果，它的结果是一个一维元组
print(res)
# print(res1)
#
# cur.scroll(0, mode=\'absolute\')  # 移动游标，绝对路径，移至最前面
# cur.scroll(0, mode=\'relative\')  # 移动游标，相对于当前位置
cur.close()
conn.close()

3. if __name__ == \'__main__\':   # 用法

a.py

print("outside: ", __name__)  # 别的文件导入这个文件，运行的结果是当前文件名tools

if __name__ == \'__main__\':  # 别人导入这个文件时，这个函数不执行，
    print(\'main:\', __name__)  # 运行当前文件时，结果是__main__

　　

b.py
# import一个文件实质是把这个文件运行了一遍
# import的文件中如果有if __name__ == \'__main__\'，导入这个文件时时不执行这个函数
import a
# 运行此文件输出结果是
a    

4. redis操作

1.　　关系型数据库：如mysql Oracle sqlserver     
        非关系型数据库：
     　　key-value格式的
     　　memcahe     # 存在内存中
     　　redis       # 存在内存中
     　　mongodb     # 数据存在磁盘中
2. 　　redis的安装：pip install redis
3. 　　

import redis
# redis 只有密码，没有用户名
# 字符串类型
r = redis.Redis(host=\'211.149.***.**\', port=6379, password=\'******\', db=1)  # 端口默认6379
# r.set(\'qxy_session\', \'201801211506\') # set数据
# print(r.get(\'qxy_session\'))  # redis取出来的数据都是bytes类型的 b\'201801211506\'
# print(r.get(\'qxy_session\').decode())  # 所以需要用decode方法转成字符串 201801211506
# r.delete(\'qxy_session\')  # 删除一个
# r.setex(\'qxy\', \'hahaha\', 20)  # 可以指定key的失效时间，单位是秒
# set get delete setex 都是针对string类型的 k - v
# 这种写法是有层级的
r.set(\'qxy:test1\', \'没交作业\')
r.set(\'qxy:test2\', \'交了作业\')
print(r.keys())  # 获取所有的key
print(r.keys(\'qxy*\'))  # 以txz开头的key
print(r.type(\'qxy:test1\'))  # 获取key的类型


# hash类型
r.hset(\'qxy_sessions\', \'q1\', \'1\')  # 插入数据
r.hset(\'qxy_sessions\', \'q2\', \'2\')
r.hset(\'qxy_sessions\', \'q3\', \'3\')
print(r.hget(\'qxy_sessions\', \'q1\').decode())  # 获取某条数据
print(r.hgetall(\'qxy_sessions\'))  # 获取hash类型中所有的类型
all_data = {}
for k,v in r.hgetall(\'qxy_sessions\').items():
    k = k.decode()
    v = v.decode()
    all_data[k] = v
print(all_data)
# hash类型没有过期时间

练习题

import redis
# 将redis中db1的数据迁移至db8中
r = redis.Redis(host=\'211.149.***.**\', port=6379, password=\'******\', db=1)
r_new = redis.Redis(host=\'211.149.***.**\', port=6379, password=\'******\', db=8)
for k in r.keys(\'\'):
    if r.type(r.keys()) == b\'string\':  # 或者用decode()
        v = r.get(k)
        r_new.set(k, v)
        print(v.decode())
    elif r.type(r.keys()) == b\'hash\':
        keys = r.hgetall(k)
        for kk, vv in keys.items():
            r_new.hset(k, kk, vv)

5. 开发接口

1. mock（模拟）接口的用处

    　　   1） 暂时代替第三方接口

    　　   2） 辅助测试：用来代替没有开发好的接口

    　　3） 查看数据

2. 需先安装flask模块：pip install flask

import flask
from conf import config
import json
from lib.tools import op_mysql
# import tools #  tools.op_mysql()

# 接口，后台服务

server = flask.Flask(__name__)


@server.route(\'/get_user\', methods=[\'get\', \'post\'])   # 这句话表示这个函数变身为接口
def get_all_user():
    sql = \'select * from users;\'
    response = op_mysql(host=config.HOST, user=config.USER, password=config.PASSWORD, db=config.DBNAME, port=config.PORT,
                        charset=\'utf8\', sql=sql)
    res = json.dumps(response, ensure_ascii=False, indent=4)
    return res


@server.route(\'/add_user\', methods=[\'post\'])
def add_users():
    user = flask.request.values.get(\'user\')
    passwd = flask.request.values.get(\'passwd\')
    print(user, passwd)
    if user and passwd:
        sql = "insert into users values(\'%s\',\'%s\');" % (user, passwd)
        op_mysql(host=config.HOST, user=config.USER, password=config.PASSWORD, db=config.DBNAME, port=config.PORT,
                 charset=\'utf8\', sql=sql)
        response = {\'code\': 200, \'msg\': \'操作成功\'}
    else:
        response = {\'code\': 503, \'msq\': \'必填参数未填\'}
    return json.dumps(response, ensure_ascii=False)


# host=\'0.0.0.0\' 代表一个局域网内的所有人都可以访问；加上debug：不需要重启服务
server.run(port=8888, host=\'0.0.0.0\', debug=True) 

在postman中访问这两个接口