Coverage for functions \ flipdare \ firestore \ backend \ exchange_rate_db.py: 100%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13 

14from google.cloud.firestore import Client as FirestoreClient 

15from flipdare.firestore._app_db import AppDb 

16from flipdare.firestore.core.db_query import DbQuery, FieldOp, WhereField 

17from flipdare.generated.model.internal.exchange_rate_model import ( 

18 ExchangeRateKeys, 

19 ExchangeRateModel, 

20) 

21from flipdare.generated.shared.stripe.stripe_currency_code import StripeCurrencyCode 

22from flipdare.generated.shared.firestore_collections import FirestoreCollections 

23from flipdare.wrapper.internal.exchange_rate_wrapper import ExchangeRateWrapper 

24 

25_EXCHANGE_RATE = FirestoreCollections.EXCHANGE_RATE.value 

26 

27__all__ = ["ExchangeRateDb"] 

28 

29_OP = FieldOp 

30_K = ExchangeRateKeys 

31 

32 

33class ExchangeRateDb(AppDb[ExchangeRateWrapper, ExchangeRateModel]): 

34 """Class for managing exchange rate-related database operations.""" 

35 

36 def __init__(self, client: FirestoreClient) -> None: 

37 super().__init__( 

38 client=client, 

39 collection_name=FirestoreCollections.EXCHANGE_RATE, 

40 model_class=ExchangeRateModel, 

41 wrapper_class=ExchangeRateWrapper, 

42 ) 

43 

44 def get_currency_id( 

45 self, 

46 base_currency: StripeCurrencyCode, 

47 target_currency: StripeCurrencyCode, 

48 ) -> str | None: 

49 """Check if an exchange rate exists for the given currency pair.""" 

50 query = DbQuery.and_( 

51 [ 

52 WhereField[_K](_K.BASE_CURRENCY, _OP.EQUAL, base_currency), 

53 WhereField[_K](_K.TARGET_CURRENCY, _OP.EQUAL, target_currency), 

54 ], 

55 limit=1, 

56 ) 

57 

58 results = query.get_query(self.client, _EXCHANGE_RATE).get() 

59 if len(results) == 0: 

60 return None 

61 

62 return str(results[0].id)