Redis搜索日期范围内的查询示例

import redis
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 存储日期数据
date = '2022-06-01'
data = 'some_data'
r.set(date, data)

import redis
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 查询日期范围
keys = r.keys('*2022-06-*')
for key in keys:
    print(key.decode(), r.get(key).decode())

import redis
import datetime
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 存储日期数据
date = '2022-06-01'
data = 'some_data'
score = datetime.datetime.strptime(date, '%Y-%m-%d').timestamp()
r.zadd('dates', {data: score})
# 查询日期范围
start_date = '2022-06-01'
end_date = '2022-06-30'
start = datetime.datetime.strptime(start_date, '%Y-%m-%d').timestamp()
end = datetime.datetime.strptime(end_date, '%Y-%m-%d').timestamp()
results = r.zrangebyscore('dates', start, end)
for result in results:
    print(result.decode())

import redis
from datetime import datetime
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 添加任务到列表
def add_task(id, deadline):
    r.zadd('tasks', {id: deadline.timestamp()})
# 查询指定日期范围内的任务
def get_tasks_in_range(start_date, end_date):
    # 转换日期格式为timestamp
    start_timestamp = datetime.strptime(start_date, '%Y-%m-%d').timestamp()
    end_timestamp = datetime.strptime(end_date, '%Y-%m-%d').timestamp()
    # 查询日期范围内的任务
    tasks = r.zrangebyscore('tasks', start_timestamp, end_timestamp)
    # 返回任务列表
    return [task.decode() for task in tasks]
# 示例代码
add_task('task1', datetime(2022, 6, 15))
add_task('task2', datetime(2022, 6, 25))
add_task('task3', datetime(2022, 7, 5))
tasks = get_tasks_in_range('2022-06-01', '2022-06-30')
print(tasks)

from flask import Flask, jsonify, request
from pymongo import MongoClient, ReturnDocument
from werkzeug.security import generate_password_hash, check_password_hash
app = Flask(__name__)
client = MongoClient('mongodb://localhost:27017/')
db = client['online_shop']
users_collection = db['users']
# 注册用户
@app.route('/register', methods=['POST'])
def register():
    username = request.json.get('username')
    password = request.json.get('password')
    # 检查用户名是否已存在
    if users_collection.find_one({'username': username}):
        return jsonify({'message': 'Username already exists'}), 400
    
    # 对密码进行哈希处理
    hashed_password = generate_password_hash(password)
    # 创建新用户
    new_user = {
        'username': username,
        'password': hashed_password
    }
    users_collection.insert_one(new_user)
    return jsonify({'message': 'User registered successfully'}), 200
# 用户登录
@app.route('/login', methods=['POST'])
def login():
    username = request.json.get('username')
    password = request.json.get('password')
    # 查找用户
    user = users_collection.find_one({'username': username})
    # 检查用户是否存在
    if not user:
        return jsonify({'message': 'User not found'}), 404
    # 检查密码是否匹配
    if not check_password_hash(user['password'], password):
        return jsonify({'message': 'Invalid password'}), 401
    return jsonify({'message': 'Login successful'}), 200
if __name__ == '__main__':
    app.run()