Python مع Redis

pip install redis

import redis
print(redis.__version__)  # يطبع رقم الإصدار

import redis

# الاتصال بـ Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# اختبار الاتصال
print(r.ping())  # True

import redis

r = redis.Redis(
    host='localhost',
    port=6379,
    password='your_password',
    db=0
)

import redis

r = redis.Redis(
    host='localhost',       # عنوان المضيف
    port=6379,              # المنفذ
    password='password',    # كلمة المرور
    db=0,                   # رقم قاعدة البيانات
    decode_responses=True,  # فك التشفير التلقائي إلى سلسلة
    socket_timeout=5,       # المهلة (ثوانٍ)
    socket_connect_timeout=5,
    retry_on_timeout=True,
    max_connections=10
)

import redis

r = redis.Redis(decode_responses=True)

# SET و GET
r.set('name', 'Alice')
print(r.get('name'))  # 'Alice'

# SET مع انتهاء صلاحية
r.set('session', 'data', ex=3600)  # تنتهي صلاحيته بعد ساعة

# MSET و MGET
r.mset({'key1': 'value1', 'key2': 'value2'})
print(r.mget('key1', 'key2'))  # ['value1', 'value2']

# INCR
r.set('counter', 0)
print(r.incr('counter'))  # 1
print(r.incrby('counter', 10))  # 11

# APPEND
r.append('name', ' Smith')
print(r.get('name'))  # 'Alice Smith'

# STRLEN
print(r.strlen('name'))  # 11

# DEL
r.delete('name')
print(r.get('name'))  # None

import redis

r = redis.Redis(decode_responses=True)

# HSET و HGET
r.hset('user:1', 'name', 'Alice')
r.hset('user:1', 'age', 25)
print(r.hget('user:1', 'name'))  # 'Alice'

# HMSET و HMGET
r.hset('user:2', mapping={'name': 'Bob', 'age': 30, 'city': 'Beijing'})
print(r.hmget('user:2', 'name', 'age'))  # ['Bob', '30']

# HGETALL
print(r.hgetall('user:2'))  # {'name': 'Bob', 'age': '30', 'city': 'Beijing'}

# HKEYS و HVALS
print(r.hkeys('user:2'))  # ['name', 'age', 'city']
print(r.hvals('user:2'))  # ['Bob', '30', 'Beijing']

# HDEL
r.hdel('user:2', 'city')

# HINCRBY
r.hincrby('user:2', 'age', 1)

import redis

r = redis.Redis(decode_responses=True)

# LPUSH و RPUSH
r.lpush('mylist', 'value1', 'value2')
r.rpush('mylist', 'value3')

# LRANGE
print(r.lrange('mylist', 0, -1))  # ['value2', 'value1', 'value3']

# LPOP و RPOP
print(r.lpop('mylist'))  # 'value2'
print(r.rpop('mylist'))  # 'value3'

# LLEN
print(r.llen('mylist'))  # 1

# LINDEX
print(r.lindex('mylist', 0))  # 'value1'

# LSET
r.lset('mylist', 0, 'new_value')

import redis

r = redis.Redis(decode_responses=True)

# SADD
r.sadd('myset', 'a', 'b', 'c')

# SMEMBERS
print(r.smembers('myset'))  # {'a', 'b', 'c'}

# SISMEMBER
print(r.sismember('myset', 'a'))  # True

# SREM
r.srem('myset', 'a')

# SCARD
print(r.scard('myset'))  # 2

# SINTER, SUNION, SDIFF
r.sadd('set1', 'a', 'b', 'c')
r.sadd('set2', 'b', 'c', 'd')
print(r.sinter('set1', 'set2'))  # {'b', 'c'}
print(r.sunion('set1', 'set2'))  # {'a', 'b', 'c', 'd'}
print(r.sdiff('set1', 'set2'))  # {'a'}

import redis

r = redis.Redis(decode_responses=True)

# ZADD
r.zadd('leaderboard', {'player1': 100, 'player2': 200, 'player3': 150})

# ZRANGE
print(r.zrange('leaderboard', 0, -1, withscores=True))
# [('player1', 100.0), ('player3', 150.0), ('player2', 200.0)]

# ZREVRANGE (من الأعلى إلى الأقل)
print(r.zrevrange('leaderboard', 0, 2, withscores=True))

# ZSCORE
print(r.zscore('leaderboard', 'player1'))  # 100.0

# ZRANK و ZREVRANK
print(r.zrank('leaderboard', 'player1'))  # 0
print(r.zrevrank('leaderboard', 'player2'))  # 0

# ZINCRBY
r.zincrby('leaderboard', 50, 'player1')

# ZREM
r.zrem('leaderboard', 'player1')

import redis

# إنشاء مجموعة اتصال
pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=100,
    decode_responses=True
)

# الحصول على اتصال من المجموعة
r = redis.Redis(connection_pool=pool)

# استخدام الاتصال
r.set('key', 'value')
print(r.get('key'))

import redis

r = redis.Redis(decode_responses=True)

# إنشاء خط أنابيب
pipe = r.pipeline()

# إضافة أوامر
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.set('key3', 'value3')
pipe.get('key1')

# تنفيذ خط الأنابيب
results = pipe.execute()
print(results)  # [True, True, True, 'value1']

# خط أنابيب مع معاملة
pipe = r.pipeline(transaction=True)
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
results = pipe.execute()

import redis

r = redis.Redis()

# نشر رسالة
r.publish('news', 'Hello World')

import redis

r = redis.Redis()

# الاشتراك في قناة
pubsub = r.pubsub()
pubsub.subscribe('news')

# الاستماع للرسائل
for message in pubsub.listen():
    if message['type'] == 'message':
        print(f"Channel: {message['channel']}")
        print(f"Message: {message['data']}")

import redis

r = redis.Redis()

pubsub = r.pubsub()
pubsub.psubscribe('news:*')

for message in pubsub.listen():
    if message['type'] == 'pmessage':
        print(f"Pattern: {message['pattern']}")
        print(f"Channel: {message['channel']}")
        print(f"Message: {message['data']}")

import redis
import json
import functools

r = redis.Redis(decode_responses=True)

def cache(expire=3600):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # إنشاء مفتاح ذاكرة مؤقتة
            cache_key = f"{func.__name__}:{args}:{kwargs}"

            # محاولة الحصول من الذاكرة المؤقتة
            cached = r.get(cache_key)
            if cached:
                return json.loads(cached)

            # تنفيذ الدالة
            result = func(*args, **kwargs)

            # تخزين في الذاكرة المؤقتة
            r.set(cache_key, json.dumps(result), ex=expire)

            return result
        return wrapper
    return decorator

@cache(expire=300)
def get_user(user_id):
    # محاكاة استعلام قاعدة بيانات
    print(f"Querying database: user_id={user_id}")
    return {'id': user_id, 'name': f'User{user_id}'}

# اختبار
print(get_user(1))  # استعلام قاعدة بيانات
print(get_user(1))  # الحصول من الذاكرة المؤقتة

import redis
import time
import uuid

r = redis.Redis()

class DistributedLock:
    def __init__(self, key, expire=10):
        self.key = f"lock:{key}"
        self.expire = expire
        self.identifier = str(uuid.uuid4())

    def acquire(self):
        # محاولة الاستحواذ على القفل
        return r.set(self.key, self.identifier, nx=True, ex=self.expire)

    def release(self):
        # تحرير القفل (باستخدام برنامج Lua النصي للذرية)
        script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        return r.eval(script, 1, self.key, self.identifier)

    def __enter__(self):
        while not self.acquire():
            time.sleep(0.1)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

# استخدام القفل
with DistributedLock('resource') as lock:
    print('Executing business logic')
    time.sleep(1)

import redis
import time

r = redis.Redis()

def rate_limit(key, limit=10, period=60):
    """
    محدد معدل
    :param key: مفتاح تحديد المعدل
    :param limit: الحد الأقصى للطلبات في النافذة الزمنية
    :param period: النافذة الزمنية (ثوانٍ)
    :return: ما إذا كان الطلب مسموحًا به
    """
    current = int(time.time())
    window_start = current - period

    # استخدام مجموعة مرتبة للنافذة المنزلقة
    pipe = r.pipeline()

    # إزالة السجلات خارج النافذة
    pipe.zremrangebyscore(key, 0, window_start)

    # عد الطلبات في النافذة الحالية
    pipe.zcard(key)

    # إضافة الطلب الحالي
    pipe.zadd(key, {str(current): current})

    # تعيين انتهاء الصلاحية
    pipe.expire(key, period)

    results = pipe.execute()
    count = results[1]

    return count < limit

# اختبار
for i in range(15):
    if rate_limit('api:user:1', limit=10, period=60):
        print(f'Request {i}: allowed')
    else:
        print(f'Request {i}: denied')

import redis
import json

r = redis.Redis(decode_responses=True)

class MessageQueue:
    def __init__(self, name):
        self.name = f"queue:{name}"

    def push(self, message):
        # إدراج في قائمة الانتظار
        r.rpush(self.name, json.dumps(message))

    def pop(self, timeout=0):
        # إخراج من قائمة الانتظار (محظور)
        result = r.blpop(self.name, timeout=timeout)
        if result:
            return json.loads(result[1])
        return None

    def size(self):
        return r.llen(self.name)

# المنتج
queue = MessageQueue('tasks')
queue.push({'task': 'send_email', 'to': 'user@example.com'})
queue.push({'task': 'generate_report', 'report_id': 123})

# المستهلك
while True:
    task = queue.pop(timeout=5)
    if task:
        print(f"Processing task: {task}")
    else:
        break

import redis
from redis.exceptions import RedisError, ConnectionError

try:
    r = redis.Redis(host='localhost', port=6379)
    r.set('key', 'value')
except ConnectionError:
    print('Connection failed')
except RedisError as e:
    print(f'Redis error: {e}')