Mejora en Obtención de Cuenta

mbilling/connector.py

+from typing import Tuple
 from sqlalchemy import create_engine, select, update
 from sqlalchemy.orm import sessionmaker
 import pathlib
@@ -7,31 +8,14 @@ from sqlalchemy.ext.automap import automap_base
 import mbilling.models as models
 import mbilling.functions as cli
 import mbilling.exceptions as mbe
-from time import sleep
-
-# CONFIGURACIÓN
-# Path de MBILLING
-mbapi_path = pathlib.Path(pathlib.Path().resolve(), 'mbilling')
-
-# Petición Correcta
-succesful = re.compile(r'^\[20\d\]')
-
-# Carga de Variables ENV
-values = dotenv_values(pathlib.Path(mbapi_path, '.env'))
-
-# Conexión a Base de Datos
-Base = automap_base()
-engine = create_engine(
-    f"mariadb+pymysql://{values['DB_USER']}:{values['DB_PASSWORD']}@{values['DB_HOST']}/{values['DB_NAME']}?charset=utf8mb4"
-)
-Base.prepare(engine, reflect=True)
-Session = sessionmaker(bind=engine)
-session = Session()
+from mbilling.db import Base, session
 
 # Carga de Tablas
 User = Base.classes.pkg_user
 Sip = Base.classes.pkg_sip
 Did = Base.classes.pkg_did
+DidDestination = Base.classes.pkg_did_destination
+DidUse = Base.classes.pkg_did_use
 
 
 def decode_rate(rate: models.Rate = models.Rate.ESTANDAR):
@@ -94,7 +78,7 @@ def get_user_account(number: str) -> User:
     return user[0] if user else None
 
 
-def get_sip_account(number: str) -> Sip:
+def get_sip_account(number: str) -> Tuple[Sip, User]:
     """Obtiene los detalles de una Cuenta SIP
 
     Args:
@@ -103,7 +87,24 @@ def get_sip_account(number: str) -> Sip:
     Returns:
         models.SipAccount: Cuenta SIP Asociada
     """
-    return session.query(Sip).filter_by(name=number).first()
+    sip = session.query(Sip).filter_by(name=number).first()
+    user = session.query(User).filter_by(id=sip.id_user).first()
+    if sip:
+        return sip, user
+    else:
+        raise mbe.SipNotFound()
+
+
+def get_sip_accounts_by_user(user: User) -> list[Sip]:
+    """Obtiene las cuentas SIP asignadas a una cuenta de facturación.
+
+    Args:
+        user (User): Cuenta de Facturación
+
+    Returns:
+        list[Sip]: Listado de Cuentas SIP
+    """
+    return session.query(Sip).filter_by(id_user=user.id).all()
 
 
 def find_user_account(number: str) -> User:
@@ -119,14 +120,11 @@ def find_user_account(number: str) -> User:
         User: Cuenta de Facturación.
     """
 
-    user = get_user_account(number)
-
-    if not user:
-        sip = get_sip_account(number)
-        if sip:
-            user = get_user_account(sip.accountcode)
-
+    sip, user = get_sip_account(number)
+    if user:
         return user
+    else:
+        raise mbe.UserNotFound()
 
 
 def get_user(number: str) -> models.User:
@@ -147,7 +145,7 @@ def get_user(number: str) -> models.User:
             select(Sip).where(Sip.id_user == user_account.id)
         ).fetchall()
         sip_accounts = [models.SipAccount.generate_from_automap(
-            sip[0]) for sip in user_sip]
+            sip[0], user.username) for sip in user_sip]
 
         user = models.User(
             username=user_account.username,
@@ -259,7 +257,7 @@ def get_dids_by_number(number: str):
     return session.query(Did).filter(Did.did.like(f'%{number}')).all()
 
 
-def update_did_owner(did: Did, new_owner: User):
+def update_did_owner(did: Did, sip: Sip):
     """Modifica el propietario de un DID
 
     Args:
@@ -267,22 +265,20 @@ def update_did_owner(did: Did, new_owner: User):
         new_owner (User): Nuevo Propietario
     """
 
-    session.execute(
-        update(Did).where(id=did.id).values(
-            id_user=new_owner.id, activated=True, reserved=True)
-    )
+    # Actualización de DID
+    did.id_user = sip.id_user
+    did.activated = True
+    did.reserved = True
 
-    session.execute(
-        update(Base.classes.pkg_did_use).where(id_did=did.id).values(
-            id_user=new_owner.id
-        )
-    )
+    # Actualización de Uso
+    destination = session.query(
+        DidDestination).filter_by(id_did=did.id).first()
+    destination.id_user = sip.id_user
+    destination.id_sip = sip.id
 
-    session.execute(
-        update(Base.classes.pkg_did_destination).where(id_did=did.id).values(
-            id_user=new_owner.id
-        )
-    )
+    # Actualización de Uso
+    use = session.query(DidUse).filter_by(id_did=did.id).first()
+    use.id_user = sip.id_user
 
     session.commit()
 
@@ -294,13 +290,13 @@ def create_or_update_did(number: str, user: User, sip: Sip = None):
             raise mbe.SipRegisteredError()
 
     for did_number in (number, f'34{number}'):
-        did = session.execute(select(Did).where(Did.did == did_number)).first()
-        if did:
-            update_did_owner(did, user)
-        else:
+        did = session.query(Did).filter_by(did=did_number).first()
+        if not did:
             cli._create_did(did_number, user.username,
                             auto_international=False)
 
+        update_did_owner(did, sip)
+
 
 def create_sip(username: str, create_sip: models.CreateSipAccount) -> models.SipAccount:
 
@@ -309,25 +305,26 @@ def create_sip(username: str, create_sip: models.CreateSipAccount) -> models.Sip
     if not user_account:
         raise mbe.UserNotFound()
 
-    if find_user_account(create_sip.cid):
-        raise mbe.SipRegisteredError()
-
+    sip = find_user_account(create_sip.cid)
+    if not sip:
         # Se generan las cuentas SIP
         cli._create_sip(create_sip.cid, create_sip.secret,
                         create_sip.host, username)
-    session.commit()
 
         # Se generan o asignan los DID
         sip = get_sip_account(create_sip.cid)
         if sip:
             create_or_update_did(create_sip.cid, user_account, sip)
-        return models.SipAccount.generate_from_automap(sip)
+            return models.SipAccount.generate_from_automap(sip, user_account.username)
         else:
             raise mbe.SipRegisteredError(
                 'La cuenta SIP no se ha generado correctamente.')
 
+    elif sip.id_user == user_account.id:
+        raise mbe.SipRegisteredError(
+            'La cuenta SIP ya está asignada al usuario.')
 
-def test_function(test_number):
+    else:
+        raise mbe.SipRegisteredError(
+            'La cuenta SIP está asignada a otro usuario.')
 
-    print(models.SipAccount.generate_from_automap(
-        get_sip_account(test_number)))