tgoop.com/pro_python_code/1010
Last Update:
π₯ 5 ΠΏΡΠΈΠΌΠ΅ΡΠΎΠ² ΠΈΡΠΏΠΎΠ»ΡΠ·ΠΎΠ²Π°Π½ΠΈΡ Redis Ρ ΠΊΠΎΠ΄ΠΎΠΌ Π½Π° Python
1. Caching
Redis ΠΌΠΎΠΆΠ½ΠΎ ΠΈΡΠΏΠΎΠ»ΡΠ·ΠΎΠ²Π°ΡΡ Π΄Π»Ρ ΠΊΡΡΠΈΡΠΎΠ²Π°Π½ΠΈΡ ΡΠ°ΡΡΠΎ ΠΈΡΠΏΠΎΠ»ΡΠ·ΡΠ΅ΠΌΡΡ
Π΄Π°Π½Π½ΡΡ
, ΡΠ½ΠΈΠΆΠ°Ρ Π½Π°Π³ΡΡΠ·ΠΊΡ Π½Π° Π²Π°ΡΠ΅ ΠΎΡΠ½ΠΎΠ²Π½ΠΎΠ΅ Ρ
ΡΠ°Π½ΠΈΠ»ΠΈΡΠ΅ Π΄Π°Π½Π½ΡΡ
. ΠΠΎΡ ΠΏΡΠΈΠΌΠ΅Ρ ΡΠΎΠ³ΠΎ, ΠΊΠ°ΠΊ ΡΠ΅Π°Π»ΠΈΠ·ΠΎΠ²Π°ΡΡ ΠΊΡΡΠΈΡΠΎΠ²Π°Π½ΠΈΠ΅ Ρ ΠΏΠΎΠΌΠΎΡΡΡ Redis Π² Pythonimport redis
# Connect to Redis
r = redis.Redis(host='localhost', port=6379, db=0)
def get_data_from_cache(key):
# Check if data exists in the cache
if r.exists(key):
# Retrieve data from the cache
data = r.get(key)
return data.decode('utf-8') # Convert bytes to string
else:
# Fetch data from the primary data source
data = fetch_data_from_source()
# Store data in the cache with a timeout of 1 hour
r.setex(key, 3600, data)
return data
2. Pub/Sub (Publish/Subscribe):
Redis ΠΏΠΎΠ΄Π΄Π΅ΡΠΆΠΈΠ²Π°Π΅Ρ ΠΏΠ°ΡΡΠ΅ΡΠ½ pub/sub, ΠΏΠΎΠ·Π²ΠΎΠ»ΡΡ Π²Π°ΠΌ ΡΠΎΠ·Π΄Π°Π²Π°ΡΡ ΡΠΈΡΡΠ΅ΠΌΡ ΠΎΠ±ΠΌΠ΅Π½Π° ΡΠΎΠΎΠ±ΡΠ΅Π½ΠΈΡΠΌΠΈ. ΠΠΎΡ ΠΏΡΠΈΠΌΠ΅Ρ:import redis
import time
# Connect to Redis
r = redis.Redis(host='localhost', port=6379, db=0)
def publish_message(channel, message):
# Publish a message to the specified channel
r.publish(channel, message)
def subscribe_channel(channel):
# Subscribe to a channel and process incoming messages
pubsub = r.pubsub()
pubsub.subscribe(channel)
for message in pubsub.listen():
print(message['data'].decode('utf-8')) # Process the received message
3. Rate Limiting:
Redis ΠΌΠΎΠΆΠ½ΠΎ ΠΈΡΠΏΠΎΠ»ΡΠ·ΠΎΠ²Π°ΡΡ Π΄Π»Ρ ΡΠ΅Π°Π»ΠΈΠ·Π°ΡΠΈΠΈ ΠΎΠ³ΡΠ°Π½ΠΈΡΠ΅Π½ΠΈΡ ΡΠΊΠΎΡΠΎΡΡΠΈ, ΡΡΠΎΠ±Ρ ΠΊΠΎΠ½ΡΡΠΎΠ»ΠΈΡΠΎΠ²Π°ΡΡ ΠΊΠΎΠ»ΠΈΡΠ΅ΡΡΠ²ΠΎ Π·Π°ΠΏΡΠΎΡΠΎΠ² ΠΈΠ»ΠΈ ΠΎΠΏΠ΅ΡΠ°ΡΠΈΠΉ Π·Π° ΠΏΠ΅ΡΠΈΠΎΠ΄ Π²ΡΠ΅ΠΌΠ΅Π½ΠΈ. ΠΡΠΈΠΌΠ΅Ρ:import redis
# Connect to Redis
r = redis.Redis(host='localhost', port=6379, db=0)
def check_rate_limit(ip_address):
# Increment the request count for the IP address
request_count = r.incr(ip_address)
# If the count exceeds the limit (e.g., 100 requests per minute), deny the request
if request_count > 100:
return False
return True
4. Session Storage:
Redis ΠΌΠΎΠΆΠ½ΠΎ ΠΈΡΠΏΠΎΠ»ΡΠ·ΠΎΠ²Π°ΡΡ Π΄Π»Ρ Ρ
ΡΠ°Π½Π΅Π½ΠΈΡ Π΄Π°Π½Π½ΡΡ
ΡΠ΅Π°Π½ΡΠ° Π² Π²Π΅Π±-ΠΏΡΠΈΠ»ΠΎΠΆΠ΅Π½ΠΈΡΡ
. ΠΡΠΈΠΌΠ΅Ρ:import redis
import uuid
# Connect to Redis
r = redis.Redis(host='localhost', port=6379, db=0)
def create_session(user_id):
# Generate a unique session ID
session_id = str(uuid.uuid4())
# Store the session data in Redis with a timeout of 30 minutes
r.setex(session_id, 1800, user_id)
return session_id
def get_user_id_from_session(session_id):
# Retrieve the user ID from the session data in Redis
user_id = r.get(session_id)
if user_id is not None:
return user_id.decode('utf-8') # Convert bytes to string
else:
return None
5. Leaderboard:
Redis ΠΌΠΎΠΆΠ½ΠΎ ΠΈΡΠΏΠΎΠ»ΡΠ·ΠΎΠ²Π°ΡΡ Π΄Π»Ρ ΡΠΎΠ·Π΄Π°Π½ΠΈΡ ΡΠ°Π±Π»ΠΈΡ Π»ΠΈΠ΄Π΅ΡΠΎΠ² ΠΈΠ»ΠΈ ΡΠ΅ΠΉΡΠΈΠ½Π³ΠΎΠ² Π½Π° ΠΎΡΠ½ΠΎΠ²Π΅ Π½Π°Π±ΡΠ°Π½Π½ΡΡ
Π±Π°Π»Π»ΠΎΠ². ΠΡΠΈΠΌΠ΅Ρ:import redis
# Connect to Redis
r = redis.Redis(host='localhost', port=6379, db=0)
def update_score(player_id, score):
# Update the score of a player
r.zadd('leaderboard', {player_id: score})
def get_leaderboard():
# Get the top 10 players from the leaderboard
leaderboard = r.zrevrange('leaderboard', 0, 9, withscores=True)
for player, score in leaderboard:
print(f"Player: {player.decode('utf-8')}, Score: {score}")
ΠΡΠΎ Π»ΠΈΡΡ Π½Π΅ΡΠΊΠΎΠ»ΡΠΊΠΎ ΠΏΡΠΈΠΌΠ΅ΡΠΎΠ² ΡΠΎΠ³ΠΎ, ΠΊΠ°ΠΊ Redis ΠΌΠΎΠΆΠ½ΠΎ ΠΈΡΠΏΠΎΠ»ΡΠ·ΠΎΠ²Π°ΡΡ Π² Python. Redis ΠΏΡΠ΅Π΄ΠΎΡΡΠ°Π²Π»ΡΠ΅Ρ ΠΌΠ½ΠΎΠΆΠ΅ΡΡΠ²ΠΎ Π΄ΡΡΠ³ΠΈΡ
ΠΌΠΎΡΠ½ΡΡ
ΡΡΠ½ΠΊΡΠΈΠΉ ΠΈ ΡΡΡΡΠΊΡΡΡ Π΄Π°Π½Π½ΡΡ
, ΠΊΠΎΡΠΎΡΡΠ΅ ΠΌΠΎΠΆΠ½ΠΎ ΠΈΡΠΏΠΎΠ»ΡΠ·ΠΎΠ²Π°ΡΡ Π² ΡΠ°Π·Π»ΠΈΡΠ½ΡΡ
ΠΏΡΠΈΠ»ΠΎΠΆΠ΅Π½ΠΈΡΡ
.
βͺGithub
@pro_python_code
BY Python RU
Share with your friend now:
tgoop.com/pro_python_code/1010