How can you implement a secure, encrypted, and scalable key-value store in Python using cryptography and redis that supports atomic operations, automatic encryption/decryption of data, concurrent access control, and seamless integration with distributed systems? Provide a concise yet comprehensive code example demonstrating advanced features such as AES-GCM encryption, transactional updates, rate limiting, and cluster-aware failover.

import redis
import asyncio
import json
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import os
import time
from typing import Dict, Any, Optional

# Configuration
REDIS_URL = "redis://localhost:6379/0"
SECRET_KEY = b"your-secure-secret-key-here-1234567890"  # Use environment variable in production
KEY_LENGTH = 32

class SecureKeyValueStore:
    def __init__(self, redis_url: str, secret_key: bytes):
        self.redis_client = redis.from_url(redis_url)
        self.fernet = Fernet(secret_key)
        self._lock = asyncio.Lock()
        self.rate_limit = {}
        
    async def _encrypt(self, data: Any) -> str:
        """Encrypt data using Fernet."""
        json_data = json.dumps(data).encode('utf-8')
        return self.fernet.encrypt(json_data).decode('utf-8')
    
    async def _decrypt(self, encrypted_data: str) -> Any:
        """Decrypt data using Fernet."""
        try:
            decrypted = self.fernet.decrypt(encrypted_data.encode('utf-8'))
            return json.loads(decrypted.decode('utf-8'))
        except Exception as e:
            logger.error(f"Decryption failed: {e}")
            return None
    
    async def set(self, key: str, value: Any, ttl: int = 300):
        """Set key-value pair with encryption and TTL."""
        encrypted_value = await self._encrypt(value)
        async with self._lock:
            await self.redis_client.setex(key, ttl, encrypted_value)
            
    async def get(self, key: str) -> Optional[Any]:
        """Get and decrypt value."""
        raw_value = await self.redis_client.get(key)
        if raw_value:
            return await self._decrypt(raw_value)
        return None
    
    async def atomic_transaction(self, operations: List[Dict]):
        """Execute atomic operations using Redis transactions."""
        pipe = self.redis_client.pipeline()
        for op in operations:
            if op['type'] == 'set':
                encrypted = await self._encrypt(op['value'])
                pipe.setex(op['key'], op.get('ttl', 300), encrypted)
            elif op['type'] == 'delete':
                pipe.delete(op['key'])
        await pipe.execute()
    
    async def rate_limited_set(self, key: str, value: Any, rate_limit: int = 10):
        """Rate-limited set operation."""
        now = time.time()
        if key not in self.rate_limit:
            self.rate_limit[key] = []
        self.rate_limit[key] = [t for t in self.rate_limit[key] if t > now - 60]
        
        if len(self.rate_limit[key]) >= rate_limit:
            raise Exception("Rate limit exceeded")
        
        self.rate_limit[key].append(now)
        await self.set(key, value)

# Example usage
async def main():
    store = SecureKeyValueStore(REDIS_URL, SECRET_KEY)
    await store.set("user:1", {"name": "John", "age": 30})
    data = await store.get("user:1")
    print(data)
    await store.atomic_transaction([
        {"type": "set", "key": "counter:1", "value": 1},
        {"type": "set", "key": "counter:2", "value": 2}
    ])
    await store.rate_limited_set("api_call", {"count": 1}, rate_limit=5)

# Run the example
asyncio.run(main())

#Python #Security #Encryption #Redis #KeyvalueStore #AtomicOperations #Concurrency #DistributedSystems #Scalability #Cryptography #AsyncIO

By: @DataScienceQ 🚀

How can you build a high-performance, fault-tolerant, and scalable web scraping framework in Python using aiohttp, selenium, asyncio, and redis to handle dynamic content, bypass anti-bot measures, and distribute crawling tasks across multiple workers? Provide a concise code example demonstrating advanced features such as rotating proxies, request rate limiting, error recovery, and distributed task queue management.

import asyncio
import aiohttp
import redis
import json
import random
from typing import Dict, Any, List
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time

# Configuration
REDIS_URL = "redis://localhost:6379/0"
PROXIES = ["https://proxy1:8080", "https://proxy2:8080"]
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}

class AsyncWebScraper:
    def __init__(self, redis_url: str):
        self.redis_client = redis.from_url(redis_url)
        self.session = None
        self.proxy = None
        
    async def setup_session(self):
        """Setup aiohttp session with proxy."""
        self.session = aiohttp.ClientSession()
        
    async def get_with_proxy(self, url: str) -> str:
        """Fetch URL with random proxy."""
        self.proxy = random.choice(PROXIES)
        headers = HEADERS.copy()
        headers["Accept"] = "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
        
        try:
            async with self.session.get(url, headers=headers, proxy=self.proxy) as response:
                return await response.text()
        except Exception as e:
            print(f"Request failed: {e}")
            return None
            
    async def scrape_with_selenium(self, url: str) -> str:
        """Scrape dynamic content using Selenium."""
        options = Options()
        options.add_argument("--headless")
        options.add_argument("--no-sandbox")
        options.add_argument("--disable-dev-shm-usage")
        
        driver = webdriver.Chrome(options=options)
        try:
            driver.get(url)
            WebDriverWait(driver, 10).until(
                EC.presence_of_element_located((By.TAG_NAME, "body"))
            )
            return driver.page_source
        finally:
            driver.quit()
            
    async def process_task(self, task_id: str, url: str):
        """Process individual scraping task."""
        # Rate limiting
        await asyncio.sleep(random.uniform(1, 3))
        
        # Try HTTP first, fallback to Selenium
        html = await self.get_with_proxy(url)
        if not html:
            html = await self.scrape_with_selenium(url)
            
        # Store result
        if html:
            await self.redis_client.set(f"result:{task_id}", html)
            
    async def worker_loop(self):
        """Worker that processes tasks from Redis queue."""
        while True:
            task = await self.redis_client.brpop("scraping_queue", timeout=5)
            if task:
                task_id, url = task[1].decode().split(":")
                await self.process_task(task_id, url)

# Example usage
async def main():
    scraper = AsyncWebScraper(REDIS_URL)
    await scraper.setup_session()
    
    # Add tasks to queue
    for i in range(5):
        await scraper.redis_client.lpush("scraping_queue", f"{i}:https://example.com")
    
    # Start worker
    await scraper.worker_loop()

asyncio.run(main())

#Python #WebScraping #AsyncIO #Selenium #Redis #ProxyRotation #FaultTolerance #DistributedSystems #DynamicContent #RateLimiting #Scalability

By: @DataScienceQ 🚀

How can you implement a hybrid AI-driven recommendation system in Python that combines collaborative filtering, content-based filtering, and real-time user behavior analysis using machine learning models (e.g., LightFM, scikit-learn) with a scalable backend powered by Redis and FastAPI to deliver personalized recommendations in real time? Provide a concise code example demonstrating advanced features such as incremental model updates, cold-start handling, A/B testing, and low-latency response generation.

import redis
import numpy as np
from fastapi import FastAPI, Depends
from typing import Dict, List, Any
from lightfm import LightFM
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import json
import asyncio

# Configuration
REDIS_URL = "redis://localhost:6379/0"
app = FastAPI()
redis_client = redis.from_url(REDIS_URL)

class HybridRecommendationSystem:
    def __init__(self):
        self.model = LightFM(no_components=30, loss='warp')
        self.user_features = {}
        self.item_features = {}
        self.tfidf = TfidfVectorizer(max_features=1000)
        
    async def update_model(self, interactions: List[Dict], items: List[Dict]):
        """Incrementally update recommendation model."""
        # Simulate training data
        n_users = len(interactions)
        n_items = len(items)
        user_ids = [i['user_id'] for i in interactions]
        item_ids = [i['item_id'] for i in interactions]
        ratings = [i['rating'] for i in interactions]
        
        # Create sparse interaction matrix
        X = np.zeros((n_users, n_items))
        for u, i, r in zip(user_ids, item_ids, ratings):
            X[u, i] = r
            
        # Update model
        self.model.fit_partial(X)
        
    async def get_recommendations(self, user_id: int, n: int = 5) -> List[int]:
        """Generate recommendations using hybrid approach."""
        # Collaborative filtering
        scores_cf = self.model.predict(user_id, np.arange(1000))
        
        # Content-based filtering
        if user_id in self.user_features:
            user_vec = np.array([self.user_features[user_id]])
            item_vecs = np.array(list(self.item_features.values()))
            scores_cb = cosine_similarity(user_vec, item_vecs)[0]
            
            # Combine scores
            combined_scores = (scores_cf + scores_cb) / 2
        else:
            combined_scores = scores_cf
            
        # Return top-N recommendations
        return np.argsort(combined_scores)[-n:][::-1].tolist()
    
    async def handle_cold_start(self, user_id: int, preferences: List[str]):
        """Handle new users with content-based recommendations."""
        # Extract features from user preferences
        tfidf_matrix = self.tfidf.fit_transform(preferences)
        user_features = tfidf_matrix.mean(axis=0).tolist()[0]
        self.user_features[user_id] = user_features
        
        # Get similar items
        return self.get_recommendations(user_id, n=10)

@app.post("/recommend")
async def recommend(user_id: int, preferences: List[str] = None):
    system = HybridRecommendationSystem()
    
    # Handle cold start
    if not preferences:
        recommendations = await system.get_recommendations(user_id)
    else:
        recommendations = await system.handle_cold_start(user_id, preferences)
    
    # Store in Redis for caching
    redis_client.set(f"rec:{user_id}", json.dumps(recommendations))
    return {"recommendations": recommendations}

# Example usage
asyncio.run(HybridRecommendationSystem().update_model(
    [{"user_id": 0, "item_id": 1, "rating": 4}],
    [{"item_id": 1, "title": "Movie A", "genre": "action"}]
))

#AI #MachineLearning #RecommendationSystems #HybridApproach #LightFM #RealTimeAI #ColdStartHandling #AandBTesting #ScalableBackend #FastAPI #Redis #Personalization

By: @DataScienceQ 🚀