Spaces:
Running
Running
File size: 5,540 Bytes
715acff a64d26e 21d8407 715acff 87444a0 715acff a64d26e 21d8407 87444a0 715acff 87444a0 715acff a64d26e 715acff a64d26e 21d8407 87444a0 715acff |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 |
from __future__ import annotations
from typing import Any, Dict
from src.agents.swap.storage import SwapStateRepository
from src.agents.dca.storage import DcaStateRepository
from src.agents.lending.storage import LendingStateRepository
from src.agents.staking.storage import StakingStateRepository
class Metadata:
def __init__(self):
self.crypto_data_agent: Dict[str, Any] = {}
self._swap_repo = SwapStateRepository.instance()
self._dca_repo = DcaStateRepository.instance()
self._lending_repo = LendingStateRepository.instance()
self._staking_repo = StakingStateRepository.instance()
def get_crypto_data_agent(self):
return self.crypto_data_agent
def set_crypto_data_agent(self, crypto_data_agent):
self.crypto_data_agent = crypto_data_agent
def get_swap_agent(self, user_id: str | None = None, conversation_id: str | None = None):
try:
return self._swap_repo.get_metadata(user_id, conversation_id)
except ValueError:
return {}
def set_swap_agent(
self,
swap_agent: Dict[str, Any] | None,
user_id: str | None = None,
conversation_id: str | None = None,
):
try:
if swap_agent:
self._swap_repo.set_metadata(user_id, conversation_id, swap_agent)
else:
self._swap_repo.clear_metadata(user_id, conversation_id)
except ValueError:
# Ignore clears when identity is missing; no actionable state to update.
return
def get_dca_agent(self, user_id: str | None = None, conversation_id: str | None = None):
try:
return self._dca_repo.get_metadata(user_id, conversation_id)
except ValueError:
return {}
def set_dca_agent(
self,
dca_agent: Dict[str, Any] | None,
user_id: str | None = None,
conversation_id: str | None = None,
):
try:
if dca_agent:
self._dca_repo.set_metadata(user_id, conversation_id, dca_agent)
else:
self._dca_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def clear_dca_agent(self, user_id: str | None = None, conversation_id: str | None = None) -> None:
try:
self._dca_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def get_dca_history(
self,
user_id: str | None = None,
conversation_id: str | None = None,
limit: int | None = None,
):
try:
return self._dca_repo.get_history(user_id, conversation_id, limit)
except ValueError:
return []
def get_swap_history(
self,
user_id: str | None = None,
conversation_id: str | None = None,
limit: int | None = None,
):
try:
return self._swap_repo.get_history(user_id, conversation_id, limit)
except ValueError:
return []
def get_lending_agent(self, user_id: str | None = None, conversation_id: str | None = None):
try:
return self._lending_repo.get_metadata(user_id, conversation_id)
except ValueError:
return {}
def set_lending_agent(
self,
lending_agent: Dict[str, Any] | None,
user_id: str | None = None,
conversation_id: str | None = None,
):
try:
if lending_agent:
self._lending_repo.set_metadata(user_id, conversation_id, lending_agent)
else:
self._lending_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def clear_lending_agent(self, user_id: str | None = None, conversation_id: str | None = None) -> None:
try:
self._lending_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def get_lending_history(
self,
user_id: str | None = None,
conversation_id: str | None = None,
limit: int | None = None,
):
try:
return self._lending_repo.get_history(user_id, conversation_id, limit)
except ValueError:
return []
def get_staking_agent(self, user_id: str | None = None, conversation_id: str | None = None):
try:
return self._staking_repo.get_metadata(user_id, conversation_id)
except ValueError:
return {}
def set_staking_agent(
self,
staking_agent: Dict[str, Any] | None,
user_id: str | None = None,
conversation_id: str | None = None,
):
try:
if staking_agent:
self._staking_repo.set_metadata(user_id, conversation_id, staking_agent)
else:
self._staking_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def clear_staking_agent(self, user_id: str | None = None, conversation_id: str | None = None) -> None:
try:
self._staking_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def get_staking_history(
self,
user_id: str | None = None,
conversation_id: str | None = None,
limit: int | None = None,
):
try:
return self._staking_repo.get_history(user_id, conversation_id, limit)
except ValueError:
return []
metadata = Metadata()
|