"""Model for vend_contact.""" import enum import uuid from typing import Optional from sqlalchemy import Column, Enum, ForeignKey, Integer, String, orm from permissions.connectors import mysql from permissions.models import ( contact as contact_model, subaccount as subaccount_model, vendor as vendor_model, ) from permissions.types import Identity, Tenant, TenantType class YNEnum(enum.Enum): """Y/N enum.""" Y = 'Y' N = 'N' class VendContact(mysql.Base): """Model for vend_contact.""" __tablename__ = 'vend_contact' id = Column(Integer, primary_key=True) # noqa: A003 contact_id = Column(Integer, ForeignKey('contact.contact_id')) contact = orm.relationship(contact_model.Contact) vendor_id = Column(Integer) subaccount_id = Column(Integer) auth0_user_id = Column(String) master = Column(Enum(YNEnum)) active = Column(Enum(YNEnum), default=YNEnum.Y) login = Column(String) @classmethod def get_by_id(cls, session: orm.session.Session, vend_contact_id: int): """Find a vend_contact by ID.""" return session.query(cls).get(vend_contact_id) @classmethod def create( cls, contact: contact_model.Contact, vendor_id: int, subaccount_id: Optional[int], email: str, auth0_user_id: str, master: YNEnum, ): """Create a vend_contact with the given attributes.""" login = generate_vend_contact_login( email=email, vendor_id=vendor_id, subaccount_id=subaccount_id, ) return cls( contact=contact, vendor_id=vendor_id, subaccount_id=subaccount_id, master=master, login=login, auth0_user_id=auth0_user_id, ) @classmethod def deactivate_by_id( cls, tx: orm.session.Session, vend_contact_id: int, ): """Deactivate a vend_contact by ID.""" tx.query(cls).filter( cls.id == vend_contact_id, cls.active == YNEnum.Y, ).update({'active': YNEnum.N}, synchronize_session=False) @classmethod def deactivate_by_vendor_uuid_for_identity( cls, tx: orm.session.Session, identity: Identity, vendor_uuid: str, ): """Deactivate a vend_contact by vendor ID.""" tx.query(cls).filter( cls.active == YNEnum.Y, cls.auth0_user_id.in_([identity.auth0_user_id, identity.id]), cls.vendor_id == vendor_model.Vendor.vendor_id, vendor_model.Vendor.vendor_uuid == vendor_uuid, ).update({'active': YNEnum.N}, synchronize_session=False) @classmethod def deactivate_by_subaccount_uuid_for_identity( cls, tx: orm.session.Session, identity: Identity, subaccount_uuid: str, ): """Deactivate a vend_contact by subaccount ID.""" tx.query(cls).filter( cls.active == YNEnum.Y, cls.auth0_user_id.in_([identity.auth0_user_id, identity.id]), cls.subaccount_id == subaccount_model.Subaccount.subaccount_id, subaccount_model.Subaccount.subaccount_uuid == subaccount_uuid, ).update({'active': YNEnum.N}, synchronize_session=False) @classmethod def update_existing_master_contact( cls, tx: orm.session.Session, vendor_id: str, ): """Update existing master contact by vendor ID.""" tx.query(cls).filter( cls.master == YNEnum.Y, cls.vendor_id == vendor_id, ).update({'master': YNEnum.N}, synchronize_session=False) @classmethod def get_next_active_for_identity_or_none(cls, session: orm.session.Session, identity: Identity): """Get the next active vend_contact for an identity if one exists, else return None.""" return ( session.query(cls) .filter( cls.active == YNEnum.Y, cls.auth0_user_id.in_([identity.auth0_user_id, identity.id]) ) .first() ) def generate_vend_contact_login(email: str, vendor_id: int, subaccount_id: Optional[int]) -> str: """Generate value for vend_contact.login column.""" if subaccount_id: login = f'{uuid.uuid4()}_{vendor_id}_{subaccount_id}_{email}' else: login = f'{uuid.uuid4()}_{vendor_id}_{email}' return login[-254:] def deactivate_by_tenant_for_identity(tx: orm.session.Session, identity: Identity, tenant: Tenant): """Deactivate a vend_contact by tenant.""" if tenant.tenant_type == TenantType.ACCOUNT: VendContact.deactivate_by_vendor_uuid_for_identity(tx, identity, tenant.tenant_uuid) elif tenant.tenant_type == TenantType.SUBACCOUNT: VendContact.deactivate_by_subaccount_uuid_for_identity(tx, identity, tenant.tenant_uuid)