RedisStorage

Описание

Предоставляет интерфейс сохранения данных в Redis

Важно

Хранение

Пример использования

from time import sleep
from atlancode_redis_storage import BaseModel, Bool, Str, Int, Float, BaseRedisStorage, Collection, List
from atlancode_redis_storage.types import NestedModel


# Описание модели пользователя с явным созданием экземпляров полей
class User(BaseModel):
    name: Str = Str()
    age: Int = Int()
    height: Float = Float()
    is_active: Bool = Bool()
    dog_names = List[Str]()
    lucky_numbers = List[Int]()
    # Nested NestedModel example
    class State(BaseModel):
        handler: Str = Str()
        step: Str = Str()

        class Config:
            key_prefix = "state"  # not used for nested key; TTL can be set here
            default_expire = None

    state = NestedModel[State]()

    class Config:
        key_prefix = "user"
        default_expire = 3600  # TTL 1 час

class Product(BaseModel):
    name: Str = Str()
    price: Float = Float()
    quantity: Int = Int()

    class Config:
        key_prefix = "product"
        default_expire = None  # TTL навсегда

# Хранилище с коллекциями
class RedisStorage(BaseRedisStorage):
    users: Collection[User, int]
    products: Collection[Product, str]

import threading
import time

def lock_user1(storage):
    with storage.users.lock(1, ttl_ms=5000):
        print("[Поток 1] Лок получен, работаю 3 секунды...")
        time.sleep(3)
        print("[Поток 1] Освобождаю лок.")

def try_lock_user1(storage):
    print("[Поток 2] Пытаюсь получить лок...")
    start = time.time()
    with storage.users.lock(1, ttl_ms=5000):
        waited = time.time() - start
        print(f"[Поток 2] Лок получен! Ждал {waited:.2f} сек.")
        print("[Поток 2] Работаю 1 секунду...")
        time.sleep(1)
        print("[Поток 2] Освобождаю лок.")


if __name__ == "__main__":
    storage = RedisStorage(
        redis_host="83.222.20.123",
        redis_port=26379,
        redis_password="YwgvIMx1uptOxo",
        redis_db=6,
        default_expire=None,  # глобальный TTL если у модели не задан
    )

    # --- Запись пользователя ---
    user1 = User(
        name="Пыхахы",
        age=10,
        height=1.9,
        is_active=True,
        dog_names=["Шарик", "Бобик"],
        lucky_numbers=[7, 42, 101],
    )
    storage.users[1] = user1

    product1 = Product(name="Хаыва", price=99999.99, quantity=1000)
    storage.products["laptop-1"] = product1
    storage.products["laptop-2"] = product1

    # --- Чтение целиком через Proxy ---
    proxy = storage.users[1]
    print("User name from proxy:", proxy.name)
    print("User age from proxy:", proxy.age)
    print("Dog names from proxy:", proxy.dog_names)
    print("Lucky numbers from proxy:", proxy.lucky_numbers)
    # --- Работа с подмоделью ---
    proxy.state.handler = "hello"  # создает/обновляет user:1:state
    print("State handler:", storage.users[1].state.handler)

    # --- Обновление одного поля ---
    proxy.age = 31
    print("Age after update:", storage.users[1].age)
    proxy.dog_names = proxy.dog_names + ["Тузик"]
    print("Dog names after update:", storage.users[1].dog_names)


    # --- Второй пользователь ---
    storage.users[2] = User(name="Alice Smith", age=28, height=5.5, dog_names=[], lucky_numbers=[])

    # --- Лок ---
    with storage.users.lock(1, ttl_ms=5000):
        print("Lock acquired on user:1")
        proxy = storage.users[1]
        proxy.age = proxy.age + 1
        sleep(1)
        proxy.age = proxy.age + 1
        print(f"Age after lock: {proxy.age}")

    # --- Итерация по пользователям ---
    print("\n--- Iterating over users ---")
    for key, user in storage.users.items():
        print(f"User key: {key}, User age: {user.age}, Lucky numbers: {user.lucky_numbers}")

    # --- Удаление ---
    #storage.users.remove(1)  # удалит также вложенные подмодели
    #storage.users.remove(2)
    print("\nUsers removed.")
    # Проверка, что ключи действительно удалены
    print("--- Checking remaining keys ---")
    for key, user in storage.users.items():
        print(f"Remaining user: {key}") # Этот цикл не должен ничего вывести
    for key, product in storage.products.items():
        print(f"Remaining product: {key}, Product price: {product.price}")
    print("Done.")

    print(f"Remaining users: {list(storage.users.keys())}")
    print(f"Remaining products: {list(storage.products.keys())}")

    print(f"Dogs User 1: {storage.users[1].dog_names}")

    print(F"Sum of Lucky Numbers User 1: {sum(storage.users[1].lucky_numbers)}")


    t1 = threading.Thread(target=lock_user1, args=(storage,))
    t2 = threading.Thread(target=try_lock_user1, args=(storage,))

    t1.start()
    time.sleep(0.5)  # чтобы t1 точно захватил лок первым
    t2.start()

    t1.join()
    t2.join()

ToDO

• Сделать исключения когда нам передает объект не с тем ключем. Ключ не того типа. И вообще изучить где у меня не возникают исключения, потому что потом искать это жопа