mirror of
https://github.com/aljazceru/nutshell.git
synced 2025-12-22 19:34:18 +01:00
e846acf946
* add websockets for quote updates * add test (not working) * wip: emit events to everyone * wip: emit events to everyone * wip, lots of things broken but invoice callback works * wip * add wip files * tests almost passing * add task * refactor nut constants * startup fix * works with old mints * wip cli * fix mypy * remove automatic invoice test now with websockets * remove comment * better logging * send back response * add rate limiter to websocket * add rate limiter to subscriptions * refactor websocket ratelimit * websocket tests * subscription kinds * doesnt start * remove circular import * update * fix mypy * move test file in test because it fails if it runs later... dunno why * adjust websocket NUT-06 settings * local import and small fix * disable websockets in CLI if "no_check" is selected * move subscription test to where it was * check proof state with callback, add tests * tests: run mint fixture per module instead of per session * subscription command name fix * test per session again * update test race conditions * fix tests * clean up * tmp * fix db issues and remove cached secrets * fix tests * blindly try pipeline * remove comments * comments
54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
from typing import Any, Dict, List, Optional
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from cashu.core.nuts import MPP_NUT, WEBSOCKETS_NUT
|
|
|
|
from ..core.base import Method, Unit
|
|
from ..core.models import Nut15MppSupport
|
|
|
|
|
|
class MintInfo(BaseModel):
|
|
name: Optional[str]
|
|
pubkey: Optional[str]
|
|
version: Optional[str]
|
|
description: Optional[str]
|
|
description_long: Optional[str]
|
|
contact: Optional[List[List[str]]]
|
|
motd: Optional[str]
|
|
nuts: Optional[Dict[int, Any]]
|
|
|
|
def __str__(self):
|
|
return f"{self.name} ({self.description})"
|
|
|
|
def supports_nut(self, nut: int) -> bool:
|
|
if self.nuts is None:
|
|
return False
|
|
return nut in self.nuts
|
|
|
|
def supports_mpp(self, method: str, unit: Unit) -> bool:
|
|
if not self.nuts:
|
|
return False
|
|
nut_15 = self.nuts.get(MPP_NUT)
|
|
if not nut_15 or not self.supports_nut(MPP_NUT):
|
|
return False
|
|
|
|
for entry in nut_15:
|
|
entry_obj = Nut15MppSupport.parse_obj(entry)
|
|
if entry_obj.method == method and entry_obj.unit == unit.name:
|
|
return True
|
|
|
|
return False
|
|
|
|
def supports_websocket_mint_quote(self, method: Method, unit: Unit) -> bool:
|
|
if not self.nuts or not self.supports_nut(WEBSOCKETS_NUT):
|
|
return False
|
|
websocket_settings = self.nuts[WEBSOCKETS_NUT]
|
|
if not websocket_settings:
|
|
return False
|
|
for entry in websocket_settings:
|
|
if entry["method"] == method.name and entry["unit"] == unit.name:
|
|
if "bolt11_mint_quote" in entry["commands"]:
|
|
return True
|
|
return False
|