from typing import Generic, List, Optional, TypeVar
from enum import Enum
from .account import Account
from .sso_account import SSOAccount
T = TypeVar("T", Account, SSOAccount)
class AccountType(Enum):
External = "External"
Internal = "Internal"
All = "All"
[docs]
class AccountFilter:
"""
Meant for use with `AccountList` object
"""
def __init__(
self,
min_access_level: int = 0,
product_type_code_filter: List = None,
product_name_filter: List[str] = None,
use_external_accounts: AccountType = AccountType.All,
**kwargs,
):
"""
:param min_access_level: Integer representing deposit/view/withdraw access: 0: No rights, 7: View, Deposit, and Withdraw Rights
:param product_type_code_filter: List of hydra_product_type_code to filter by
:param product_name_filter: List of hydra product names to filter by
:param account_types: AccountType Enum value: External, Internal, or All
"""
# For backwards compatibility
if "product_code_filter" in kwargs:
product_type_code_filter = kwargs["product_code_filter"]
self.min_access_level = min_access_level
self.product_type_code_filter = product_type_code_filter
self.account_types = use_external_accounts
self.product_name_filter = product_name_filter
[docs]
class AccountList(Generic[T]):
"""
Object for holding multiple Account objects. Provides search options.
"""
def __init__(
self,
accounts: Optional[List[T]] = None,
account_filter: Optional[AccountFilter] = None,
):
if accounts is None:
accounts = []
self.accounts = accounts
self.account_filter = account_filter
self.accounts_by_id = {x.host_acct_id: x for x in accounts}
@property
def filtered_accounts(self) -> List[T]:
"""self.accounts after self.account_filter is applied"""
if self.account_filter:
return [x for x in self.accounts if self._matches_config(x)]
return self.accounts
@property
def _account_dict(self) -> dict:
return {x.host_acct_id: x for x in self.filtered_accounts}
def append(self, account: T):
if account.host_acct_id in self.accounts_by_id:
insert_index = [
i
for i, x in enumerate(self.accounts)
if x.host_acct_id == account.host_acct_id
][0]
self.accounts[insert_index] = account
else:
self.accounts.append(account)
self.accounts_by_id[account.host_acct_id] = account
def get_by_id(self, host_account_id: int) -> T:
if self.account_filter:
return self._account_dict.get(host_account_id)
return self.accounts_by_id.get(host_account_id)
[docs]
def get_by_account_number(self, acct_num: str) -> T:
"""
Return account with given acct_number_internal_unmasked
"""
for acct in self.filtered_accounts:
if acct.acct_number_internal_unmasked == acct_num:
return acct
[docs]
def filter_by_cif(self, cif: str) -> List[T]:
"""
Return accounts with given cif
"""
return list(set([x for x in self.filtered_accounts if x.cif == cif]))
@property
def filtered_cifs(self) -> List[str]:
"""
CIFs of accounts in self.filtered_accounts
"""
return list(set([x.cif for x in self.filtered_accounts]))
@property
def filtered_account_numbers(self) -> List[str]:
"""
Acct Numbers of accounts in self.filtered_accounts
"""
return list(
set([x.acct_number_internal_unmasked for x in self.filtered_accounts])
)
[docs]
def get_host_acct_id_list(self, cif=None, account_number=None):
"""
:param cif: If provided, will limit the result by cif
"""
account_list = self.filtered_accounts
if cif:
account_list = [x for x in account_list if x.cif == cif]
if account_number:
account_list = [
x
for x in account_list
if x.acct_number_internal_unmasked == account_number
]
return list(set([x.host_acct_id for x in account_list]))
def _matches_config(self, account: Account) -> bool:
if self.account_filter and isinstance(account, Account):
if self.account_filter.account_types == AccountType.Internal:
if account.is_external_acct is True:
return False
elif self.account_filter.account_types == AccountType.External:
if account.is_external_acct is False:
return False
if self.account_filter.product_type_code_filter:
if (
account.hydra_product_type_code
not in self.account_filter.product_type_code_filter
):
return False
if self.account_filter.product_name_filter:
if (
account.hydra_product_name
not in self.account_filter.product_name_filter
):
return False
if (
isinstance(account.access, int)
and (self.account_filter.min_access_level & account.access)
!= self.account_filter.min_access_level
):
return False
return True
def __getitem__(self, i) -> T:
return self.accounts[i]
def __len__(self):
return len(self.accounts)
def __repr__(self):
return "<Object AccountList: {} accounts>".format(len(self))