Source code for src.escrow.escrow_offer

# Copyright (C) 2019  alfred richardsn
#
# This file is part of TellerBot.
#
# TellerBot is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with TellerBot.  If not, see <https://www.gnu.org/licenses/>.
import typing
from dataclasses import dataclass

from bson.decimal128 import Decimal128
from bson.objectid import ObjectId

from src.database import database


[docs]def asdict(instance): """Represent class instance as dictionary excluding None values.""" return {key: value for key, value in instance.__dict__.items() if value is not None}
[docs]@dataclass class EscrowOffer: """Class used to represent escrow offer. Attributes correspond to fields in database document. """ #: Primary key value of offer document. _id: ObjectId #: Primary key value of corresponding order document. order: ObjectId #: Currency which order creator wants to buy. buy: str #: Currency which order creator wants to sell. sell: str #: Type of offer. Field of currency which is held during exchange. type: str # noqa: A003 #: Currency which is held during exchange. escrow: str #: Unix time stamp of offer creation. time: float #: Object representing initiator of escrow. init: typing.Mapping[str, typing.Any] #: Object representing counteragent of escrow. counter: typing.Mapping[str, typing.Any] #: Telegram ID of user required to send message to bot. pending_input_from: typing.Optional[int] = None #: Temporary field of currency in which user is sending amount. sum_currency: typing.Optional[str] = None #: Amount in ``buy`` currency. sum_buy: typing.Optional[Decimal128] = None #: Amount in ``sell`` currency. sum_sell: typing.Optional[Decimal128] = None #: Amount of held currency with agreed fee added. sum_fee_up: typing.Optional[Decimal128] = None #: Amount of held currency with agreed fee substracted. sum_fee_down: typing.Optional[Decimal128] = None #: Amount of insured currency. insured: typing.Optional[Decimal128] = None #: Unix time stamp of counteragent first reaction to sent offer. react_time: typing.Optional[float] = None #: Unix time stamp since which transaction should be checked. transaction_time: typing.Optional[float] = None #: Unix time stamp of offer cancellation. cancel_time: typing.Optional[float] = None #: Bank of fiat currency. bank: typing.Optional[str] = None #: Required memo in blockchain transaction. memo: typing.Optional[str] = None #: ID of verified transaction. trx_id: typing.Optional[str] = None #: True if non-escrow token sender hasn't confirmed their transfer. unsent: typing.Optional[bool] = None def __getitem__(self, key: str) -> typing.Any: """Allow to use class as dictionary.""" return asdict(self)[key]
[docs] async def insert_document(self) -> None: """Convert self to document and insert to database.""" await database.escrow.insert_one(asdict(self))
[docs] async def update_document(self, update) -> None: """Update corresponding document in database. :param update: Document with update operators or aggregation pipeline sent to MongoDB. """ await database.escrow.update_one({"_id": self._id}, update)
[docs] async def delete_document(self) -> None: """Archive and delete corresponding document in database.""" await database.escrow_archive.insert_one(asdict(self)) await database.escrow.delete_one({"_id": self._id})