Coverage for dibbler / models / Transaction.py: 94%
123 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-01-25 14:26 +0000
1from __future__ import annotations
3from datetime import datetime
4from typing import TYPE_CHECKING, Self
6from sqlalchemy import (
7 CheckConstraint,
8 DateTime,
9 ForeignKey,
10 Integer,
11 Text,
12 and_,
13 column,
14 func,
15 or_,
16)
17from sqlalchemy.orm import (
18 Mapped,
19 mapped_column,
20 relationship,
21)
22from sqlalchemy.orm.collections import (
23 InstrumentedDict,
24 InstrumentedList,
25 InstrumentedSet,
26)
27from sqlalchemy.sql.schema import Index
29from .Base import Base
30from .TransactionType import TransactionType, TransactionTypeSQL
32if TYPE_CHECKING:
33 from .Product import Product
34 from .User import User
36# NOTE: these only matter when there are no adjustments made in the database.
37DEFAULT_INTEREST_RATE_PERCENT = 100
38DEFAULT_PENALTY_THRESHOLD = -100
39DEFAULT_PENALTY_MULTIPLIER_PERCENT = 200
41_DYNAMIC_FIELDS: set[str] = {
42 "amount",
43 "interest_rate_percent",
44 "joint_transaction_id",
45 "penalty_multiplier_percent",
46 "penalty_threshold",
47 "per_product",
48 "product_count",
49 "product_id",
50 "transfer_user_id",
51}
53EXPECTED_FIELDS: dict[TransactionType, set[str]] = {
54 TransactionType.ADD_PRODUCT: {"amount", "per_product", "product_count", "product_id"},
55 TransactionType.ADJUST_BALANCE: {"amount"},
56 TransactionType.ADJUST_INTEREST: {"interest_rate_percent"},
57 TransactionType.ADJUST_PENALTY: {"penalty_multiplier_percent", "penalty_threshold"},
58 TransactionType.ADJUST_STOCK: {"product_count", "product_id"},
59 TransactionType.BUY_PRODUCT: {"product_count", "product_id"},
60 TransactionType.JOINT: {"product_count", "product_id"},
61 TransactionType.JOINT_BUY_PRODUCT: {"joint_transaction_id"},
62 TransactionType.THROW_PRODUCT: {"product_count", "product_id"},
63 TransactionType.TRANSFER: {"amount", "transfer_user_id"},
64}
66assert all(x <= _DYNAMIC_FIELDS for x in EXPECTED_FIELDS.values()), (
67 "All expected fields must be part of _DYNAMIC_FIELDS."
68)
71def _transaction_type_field_constraints(
72 transaction_type: TransactionType,
73 expected_fields: set[str],
74) -> CheckConstraint:
75 unexpected_fields = _DYNAMIC_FIELDS - expected_fields
77 return CheckConstraint(
78 or_(
79 column("type") != transaction_type.value,
80 and_(
81 *[column(field).is_not(None) for field in expected_fields],
82 *[column(field).is_(None) for field in unexpected_fields],
83 ),
84 ),
85 name=f"trx_type_{transaction_type.value}_expected_fields",
86 )
89class Transaction(Base):
90 __tablename__ = "trx"
91 __table_args__ = (
92 *[
93 _transaction_type_field_constraints(transaction_type, expected_fields)
94 for transaction_type, expected_fields in EXPECTED_FIELDS.items()
95 ],
96 CheckConstraint(
97 or_(
98 column("type") != TransactionType.TRANSFER.value,
99 column("user_id") != column("transfer_user_id"),
100 ),
101 name="trx_type_transfer_no_self_transfers",
102 ),
103 CheckConstraint(
104 func.coalesce(column("product_count"), 1) != 0,
105 name="trx_product_count_non_zero",
106 ),
107 CheckConstraint(
108 func.coalesce(column("penalty_multiplier_percent"), 100) >= 100,
109 name="trx_penalty_multiplier_percent_min_100",
110 ),
111 CheckConstraint(
112 func.coalesce(column("interest_rate_percent"), 0) >= 0,
113 name="trx_interest_rate_percent_non_negative",
114 ),
115 CheckConstraint(
116 func.coalesce(column("amount"), 1) != 0,
117 name="trx_amount_non_zero",
118 ),
119 CheckConstraint(
120 func.coalesce(column("per_product"), 1) > 0,
121 name="trx_per_product_positive",
122 ),
123 CheckConstraint(
124 func.coalesce(column("penalty_threshold"), 0) <= 0,
125 name="trx_penalty_threshold_max_0",
126 ),
127 CheckConstraint(
128 or_(
129 column("joint_transaction_id").is_(None),
130 column("joint_transaction_id") != column("id"),
131 ),
132 name="trx_joint_transaction_id_not_self",
133 ),
135 # Speed up product stock calculation
136 Index("ix__transaction__product_id_type_time", "product_id", "type", "time"),
138 # Speed up product owner calculation
139 Index("ix__transaction__user_id_product_time", "user_id", "product_id", "time"),
141 # Speed up user transaction list / credit calculation
142 Index("ix__transaction__user_id_time", "user_id", "time"),
143 )
145 id: Mapped[int] = mapped_column(Integer, primary_key=True)
146 """
147 A unique identifier for the transaction.
149 Not used for anything else than identifying the transaction in the database.
150 """
152 time: Mapped[datetime] = mapped_column(DateTime, index=True)
153 """
154 The time when the transaction took place.
156 This is used to order transactions chronologically, and to calculate
157 all kinds of state.
158 """
160 message: Mapped[str | None] = mapped_column(Text, nullable=True)
161 """
162 A message that can be set by the user to describe the reason
163 behind the transaction (or potentially a place to write som fan fiction).
165 This is not used for any calculations, but can be useful for debugging.
166 """
168 type_: Mapped[TransactionType] = mapped_column(TransactionTypeSQL, name="type", index=True)
169 """
170 Which type of transaction this is.
172 The type determines which fields are expected to be set.
173 """
175 amount: Mapped[int | None] = mapped_column(Integer)
176 """
177 This field means different things depending on the transaction type:
179 - `ADD_PRODUCT`: The real amount spent on the products.
181 - `ADJUST_BALANCE`: The amount of credit to add or subtract from the user's balance.
183 - `TRANSFER`: The amount of balance to transfer to another user.
184 """
186 per_product: Mapped[int | None] = mapped_column(Integer)
187 """
188 If adding products, how much is each product worth
190 Note that this is distinct from the total amount of the transaction,
191 because this gets rounded up to the nearest integer, while the total amount
192 that the user paid in the store would be stored in the `amount` field.
193 """
195 user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), index=True)
196 """The user who performs the transaction. See `user` for more details."""
197 user: Mapped[User] = relationship(
198 lazy="joined",
199 foreign_keys=[user_id],
200 )
201 """
202 The user who performs the transaction.
204 For some transaction types, like `TRANSFER` and `ADD_PRODUCT`, this is a
205 functional field with "real world consequences" for price calculations.
207 For others, like `ADJUST_PENALTY` and `ADJUST_STOCK`, this is just a record of who
208 performed the transaction, and does not affect any state calculations.
210 In the case of `JOINT` transactions, this is the user who initiated the joint transaction.
211 """
213 joint_transaction_id: Mapped[int | None] = mapped_column(
214 ForeignKey("trx.id"),
215 index=True,
216 )
217 """
218 An optional ID to group multiple transactions together as part of a joint transaction.
220 This is used for `JOINT` and `JOINT_BUY_PRODUCT` transactions, where multiple users
221 are involved in a single transaction.
222 """
223 joint_transaction: Mapped[Transaction | None] = relationship(
224 lazy="joined",
225 foreign_keys=[joint_transaction_id],
226 )
227 """
228 The joint transaction that this transaction is part of, if any.
229 """
231 # Receiving user when moving credit from one user to another
232 transfer_user_id: Mapped[int | None] = mapped_column(ForeignKey("user.id"), index=True)
233 """The user who receives money in a `TRANSFER` transaction."""
234 transfer_user: Mapped[User | None] = relationship(
235 lazy="joined",
236 foreign_keys=[transfer_user_id],
237 )
238 """The user who receives money in a `TRANSFER` transaction."""
240 # The product that is either being added or bought
241 product_id: Mapped[int | None] = mapped_column(ForeignKey("product.id"), index=True)
242 """The product being added or bought."""
243 product: Mapped[Product | None] = relationship(lazy="joined")
244 """The product being added or bought."""
246 # The amount of products being added or bought
247 product_count: Mapped[int | None] = mapped_column(Integer)
248 """
249 The amount of products being added or bought.
251 This is always relative to the existing stock.
253 - `ADD_PRODUCT` increases the stock by this amount.
255 - `BUY_PRODUCT` decreases the stock by this amount.
257 - `ADJUST_STOCK` increases or decreases the stock by this amount,
258 depending on whether the amount is positive or negative.
259 """
261 penalty_threshold: Mapped[int | None] = mapped_column(Integer, nullable=True)
262 """
263 On `ADJUST_PENALTY` transactions, this is the threshold in krs for when the user
264 should start getting penalized for low credit.
266 See also `penalty_multiplier`.
267 """
269 penalty_multiplier_percent: Mapped[int | None] = mapped_column(Integer, nullable=True)
270 """
271 On `ADJUST_PENALTY` transactions, this is the multiplier for the amount of
272 money the user has to pay when they have too low credit.
274 The multiplier is a percentage, so `100` means the user has to pay the full
275 price of the product, `200` means they have to pay double, etc.
277 See also `penalty_threshold`.
278 """
280 interest_rate_percent: Mapped[int | None] = mapped_column(Integer, nullable=True)
281 """
282 On `ADJUST_INTEREST` transactions, this is the interest rate in percent
283 that the user has to pay on their balance.
285 The interest rate is a percentage, so `100` means the user has to pay the full
286 price of the product, `200` means they have to pay double, etc.
287 """
289 economy_spec_version: Mapped[int] = mapped_column(Integer, default=1)
290 """
291 The version of the economy specification that this transaction adheres to.
293 This is used to handle changes in the economy rules over time.
294 """
296 def __init__(
297 self: Self,
298 type_: TransactionType,
299 user_id: int,
300 amount: int | None = None,
301 interest_rate_percent: int | None = None,
302 joint_transaction_id: int | None = None,
303 message: str | None = None,
304 penalty_multiplier_percent: int | None = None,
305 penalty_threshold: int | None = None,
306 per_product: int | None = None,
307 product_count: int | None = None,
308 product_id: int | None = None,
309 time: datetime | None = None,
310 transfer_user_id: int | None = None,
311 ) -> None:
312 """
313 Please do not call this constructor directly, use the factory methods instead.
314 """
315 if time is None:
316 time = datetime.now()
318 self.amount = amount
319 self.interest_rate_percent = interest_rate_percent
320 self.joint_transaction_id = joint_transaction_id
321 self.message = message
322 self.penalty_multiplier_percent = penalty_multiplier_percent
323 self.penalty_threshold = penalty_threshold
324 self.per_product = per_product
325 self.product_count = product_count
326 self.product_id = product_id
327 self.time = time
328 self.transfer_user_id = transfer_user_id
329 self.type_ = type_
330 self.user_id = user_id
332 self._validate_by_transaction_type()
334 def _validate_by_transaction_type(self: Self) -> None:
335 """
336 Validates the transaction's fields based on its type.
337 Raises `ValueError` if the transaction is invalid.
338 """
339 if self.amount == 0: 339 ↛ 340line 339 didn't jump to line 340 because the condition on line 339 was never true
340 raise ValueError("Amount must not be zero.")
342 for field in EXPECTED_FIELDS[self.type_]:
343 if getattr(self, field) is None: 343 ↛ 344line 343 didn't jump to line 344 because the condition on line 343 was never true
344 raise ValueError(f"{field} must not be None for {self.type_.value} transactions.")
346 for field in _DYNAMIC_FIELDS - EXPECTED_FIELDS[self.type_]:
347 if getattr(self, field) is not None: 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true
348 raise ValueError(f"{field} must be None for {self.type_.value} transactions.")
350 if self.per_product is not None and self.per_product <= 0: 350 ↛ 351line 350 didn't jump to line 351 because the condition on line 350 was never true
351 raise ValueError("per_product must be greater than zero.")
353 if (
354 self.per_product is not None
355 and self.product_count is not None
356 and self.amount is not None
357 and self.amount > self.per_product * self.product_count
358 ):
359 raise ValueError(
360 "The real amount of the transaction must be less than the total value of the products."
361 )
363 # TODO: improve printing further
365 def __repr__(self) -> str:
366 sort_order = [
367 "id",
368 "time",
369 ]
371 columns = ", ".join(
372 f"{k}={repr(v)}"
373 for k, v in sorted(
374 self.__dict__.items(),
375 key=lambda item: chr(sort_order.index(item[0]))
376 if item[0] in sort_order
377 else item[0],
378 )
379 if not any(
380 [
381 k == "type_",
382 (k == "message" and v is None),
383 k.startswith("_"),
384 # Ensure that we don't try to print out the entire list of
385 # relationships, which could create an infinite loop
386 isinstance(v, Base),
387 isinstance(v, InstrumentedList),
388 isinstance(v, InstrumentedSet),
389 isinstance(v, InstrumentedDict),
390 *[k in (_DYNAMIC_FIELDS - EXPECTED_FIELDS[self.type_])],
391 ]
392 )
393 )
394 return f"{self.type_.upper()}({columns})"
396 ###################
397 # FACTORY METHODS #
398 ###################
400 @classmethod
401 def adjust_balance(
402 cls: type[Self],
403 amount: int,
404 user_id: int,
405 time: datetime | None = None,
406 message: str | None = None,
407 ) -> Self:
408 """
409 Convenience constructor for creating an `ADJUST_BALANCE` transaction.
411 Should NOT be used directly in the application code; use the various queries instead.
412 """
413 return cls(
414 time=time,
415 type_=TransactionType.ADJUST_BALANCE,
416 amount=amount,
417 user_id=user_id,
418 message=message,
419 )
421 @classmethod
422 def adjust_interest(
423 cls: type[Self],
424 interest_rate_percent: int,
425 user_id: int,
426 time: datetime | None = None,
427 message: str | None = None,
428 ) -> Self:
429 """
430 Convenience constructor for creating an `ADJUST_INTEREST` transaction.
432 Note that the `interest_rate_percent` is absolute, not relative to the previous interest rate.
434 Should NOT be used directly in the application code; use the various queries instead.
435 """
437 return cls(
438 time=time,
439 type_=TransactionType.ADJUST_INTEREST,
440 interest_rate_percent=interest_rate_percent,
441 user_id=user_id,
442 message=message,
443 )
445 @classmethod
446 def adjust_penalty(
447 cls: type[Self],
448 penalty_multiplier_percent: int,
449 penalty_threshold: int,
450 user_id: int,
451 time: datetime | None = None,
452 message: str | None = None,
453 ) -> Self:
454 """
455 Convenience constructor for creating an `ADJUST_PENALTY` transaction.
457 Note that both `penalty_multiplier_percent` and `penalty_threshold` are absolute,
458 not relative to the previous settings.
460 Should NOT be used directly in the application code; use the various queries instead.
461 """
462 return cls(
463 time=time,
464 type_=TransactionType.ADJUST_PENALTY,
465 penalty_multiplier_percent=penalty_multiplier_percent,
466 penalty_threshold=penalty_threshold,
467 user_id=user_id,
468 message=message,
469 )
471 @classmethod
472 def adjust_stock(
473 cls: type[Self],
474 user_id: int,
475 product_id: int,
476 product_count: int,
477 time: datetime | None = None,
478 message: str | None = None,
479 ) -> Self:
480 """
481 Convenience constructor for creating an `ADJUST_STOCK` transaction.
483 Should NOT be used directly in the application code; use the various queries instead.
484 """
485 return cls(
486 time=time,
487 type_=TransactionType.ADJUST_STOCK,
488 user_id=user_id,
489 product_id=product_id,
490 product_count=product_count,
491 message=message,
492 )
494 @classmethod
495 def add_product(
496 cls: type[Self],
497 amount: int,
498 user_id: int,
499 product_id: int,
500 per_product: int,
501 product_count: int,
502 time: datetime | None = None,
503 message: str | None = None,
504 ) -> Self:
505 """
506 Convenience constructor for creating an `ADD_PRODUCT` transaction.
508 Should NOT be used directly in the application code; use the various queries instead.
509 """
510 return cls(
511 time=time,
512 type_=TransactionType.ADD_PRODUCT,
513 amount=amount,
514 user_id=user_id,
515 product_id=product_id,
516 per_product=per_product,
517 product_count=product_count,
518 message=message,
519 )
521 @classmethod
522 def buy_product(
523 cls: type[Self],
524 user_id: int,
525 product_id: int,
526 product_count: int,
527 time: datetime | None = None,
528 message: str | None = None,
529 ) -> Self:
530 """
531 Convenience constructor for creating a `BUY_PRODUCT` transaction.
533 Should NOT be used directly in the application code; use the various queries instead.
534 """
535 return cls(
536 time=time,
537 type_=TransactionType.BUY_PRODUCT,
538 user_id=user_id,
539 product_id=product_id,
540 product_count=product_count,
541 message=message,
542 )
544 @classmethod
545 def joint(
546 cls: type[Self],
547 user_id: int,
548 product_id: int,
549 product_count: int,
550 time: datetime | None = None,
551 message: str | None = None,
552 ) -> Self:
553 """
554 Convenience constructor for creating a `JOINT` transaction.
556 Should NOT be used directly in the application code; use the various queries instead.
557 """
558 return cls(
559 time=time,
560 type_=TransactionType.JOINT,
561 user_id=user_id,
562 product_id=product_id,
563 product_count=product_count,
564 message=message,
565 )
567 @classmethod
568 def joint_buy_product(
569 cls: type[Self],
570 joint_transaction_id: int,
571 user_id: int,
572 time: datetime | None = None,
573 message: str | None = None,
574 ) -> Self:
575 """
576 Convenience constructor for creating a `JOINT_BUY_PRODUCT` transaction.
578 Should NOT be used directly in the application code; use the various queries instead.
579 """
580 return cls(
581 time=time,
582 type_=TransactionType.JOINT_BUY_PRODUCT,
583 joint_transaction_id=joint_transaction_id,
584 user_id=user_id,
585 message=message,
586 )
588 @classmethod
589 def transfer(
590 cls: type[Self],
591 amount: int,
592 user_id: int,
593 transfer_user_id: int,
594 time: datetime | None = None,
595 message: str | None = None,
596 ) -> Self:
597 """
598 Convenience constructor for creating a `TRANSFER` transaction.
600 Should NOT be used directly in the application code; use the various queries instead.
601 """
602 return cls(
603 time=time,
604 type_=TransactionType.TRANSFER,
605 amount=amount,
606 user_id=user_id,
607 transfer_user_id=transfer_user_id,
608 message=message,
609 )
611 @classmethod
612 def throw_product(
613 cls: type[Self],
614 user_id: int,
615 product_id: int,
616 product_count: int,
617 time: datetime | None = None,
618 message: str | None = None,
619 ) -> Self:
620 """
621 Convenience constructor for creating a `THROW_PRODUCT` transaction.
623 Should NOT be used directly in the application code; use the various queries instead.
624 """
625 return cls(
626 time=time,
627 type_=TransactionType.THROW_PRODUCT,
628 user_id=user_id,
629 product_id=product_id,
630 product_count=product_count,
631 message=message,
632 )