Add remaining web services

Author: ivanalemunioz

afip/afip.py

@@ -1,90 +1,96 @@
 import http.client
 import json
 
-from .webservice import WebService
-from .electronicbilling import ElectronicBilling
+from .web_service import WebService
+from .electronic_billing import ElectronicBilling
+from .register_inscription_proof import RegisterInscriptionProof
+from .register_scope_ten import RegisterScopeTen
+from .register_scope_thirteen import RegisterScopeThirteen
 
 class Afip:
-    sdk_version_number = '1.0.0'
-
-    def __init__(self, options: dict):
-        if not(options.get("CUIT")):
-            raise Exception("CUIT field is required in options")
-
-        self.CUIT: int = options.get("CUIT")
-        self.production: bool = options.get("production") if options.get("production") == True else False
-        self.environment: str = "prod" if self.production == True else "dev"
-        self.cert: str = options.get("cert")
-        self.key: str = options.get("key")
-        self.access_token: str = options.get("access_token")
+  sdk_version_number = '1.0.0'
+
+  def __init__(self, options: dict):
+    if not(options.get("CUIT")):
+      raise Exception("CUIT field is required in options")
+
+    self.CUIT: int = options.get("CUIT")
+    self.production: bool = options.get("production") if options.get("production") == True else False
+    self.environment: str = "prod" if self.production == True else "dev"
+    self.cert: str = options.get("cert")
+    self.key: str = options.get("key")
+    self.access_token: str = options.get("access_token")
+  
+    self.ElectronicBilling = ElectronicBilling(self)
+    self.RegisterInscriptionProof = RegisterInscriptionProof(self)
+    self.RegisterScopeTen = RegisterScopeTen(self)
+    self.RegisterScopeThirteen = RegisterScopeThirteen(self)
+
+
+  # Gets token authorization for an AFIP Web Service
+  #
+  # If force is true it forces to create a new TA
+  def getServiceTA(self, service: str, force: bool = False) -> dict:
+    conn = http.client.HTTPSConnection("app.afipsdk.com")
+
+    payload = {
+      "environment": self.environment,
+      "tax_id": self.CUIT,
+      "wsid": service,
+      "force_create": force
+    }
+
+    if self.cert: payload["cert"] = self.cert
+    if self.key: payload["key"] = self.key
+
+    headers = {
+      "Content-Type": "application/json",
+      "sdk-version-number": self.sdk_version_number,
+      "sdk-library": "python",
+      "sdk-environment": self.environment
+    }
+
+    if self.access_token: headers["Authorization"] = "Bearer %s" % self.access_token
+
+    conn.request("POST", "/api/v1/afip/auth", json.dumps(payload), headers)
+
+    res = conn.getresponse()
 
-        self.ElectronicBilling = ElectronicBilling(self)
+    data = res.read()
 
+    if res.getcode() >= 400:
+      raise Exception(data.decode("utf-8"))
 
-    # Gets token authorization for an AFIP Web Service
-    #
-    # If force is true it forces to create a new TA
-    def getServiceTA(self, service: str, force: bool = False) -> dict:
-        conn = http.client.HTTPSConnection("app.afipsdk.com")
+    return json.loads(data.decode("utf-8"))
+  
+  # Get last request and last response XML
+  def getLastRequestXML(self) -> dict:
+    conn = http.client.HTTPSConnection("app.afipsdk.com")
 
-        payload = {
-            "environment": self.environment,
-            "tax_id": self.CUIT,
-            "wsid": service,
-            "force_create": force
-        }
+    headers = {
+      "Content-Type": "application/json",
+      "sdk-version-number": self.sdk_version_number,
+      "sdk-library": "python",
+      "sdk-environment": self.environment
+    }
 
-        if self.cert: payload["cert"] = self.cert
-        if self.key: payload["key"] = self.key
+    if self.access_token: headers["Authorization"] = "Bearer %s" % self.access_token
 
-        headers = {
-            "Content-Type": "application/json",
-            "sdk-version-number": self.sdk_version_number,
-            "sdk-library": "python",
-            "sdk-environment": self.environment
-        }
+    conn.request("GET", "/api/v1/afip/requests/last-xml", None, headers)
 
-        if self.access_token: headers["Authorization"] = "Bearer %s" % self.access_token
-
-        conn.request("POST", "/api/v1/afip/auth", json.dumps(payload), headers)
-
-        res = conn.getresponse()
-        
-        data = res.read()
-
-        if res.getcode() >= 400:
-            raise Exception(data.decode("utf-8"))
-
-        return json.loads(data.decode("utf-8"))
+    res = conn.getresponse()
 
-    # Get last request and last response XML
-    def getLastRequestXML(self) -> dict:
-        conn = http.client.HTTPSConnection("app.afipsdk.com")
-
-        headers = {
-            "Content-Type": "application/json",
-            "sdk-version-number": self.sdk_version_number,
-            "sdk-library": "python",
-            "sdk-environment": self.environment
-        }
+    data = res.read()
 
-        if self.access_token: headers["Authorization"] = "Bearer %s" % self.access_token
+    if res.getcode() >= 400:
+      raise Exception(data.decode("utf-8"))
 
-        conn.request("GET", "/api/v1/afip/requests/last-xml", None, headers)
-
-        res = conn.getresponse()
-        
-        data = res.read()
-
-        if res.getcode() >= 400:
-            raise Exception(data.decode("utf-8"))
-
-        return json.loads(data.decode("utf-8"))
-    
+    return json.loads(data.decode("utf-8"))
+  
 
-    # Create generic Web Service
-    def webService(self, service, options: dict = {}) -> WebService:
-        options['service'] = service
-        options['generic'] = True
+  # Create generic Web Service
+  def webService(self, service, options: dict = {}) -> WebService:
+    options['service'] = service
+    options['generic'] = True
 
-        return WebService(self, options)
+    return WebService(self, options)

afip/electronicbilling.py afip/electronic_billing.pyafip/electronicbilling.py renamed to afip/electronic_billing.py

@@ -2,7 +2,7 @@
 import json
 import re
 
-from .webservice import WebService
+from .web_service import WebService
 
 class ElectronicBilling(WebService):
   soapv12 = True

afip/register_inscription_proof.py

@@ -0,0 +1,56 @@
+from .web_service import WebService
+
+class RegisterInscriptionProof(WebService):
+  soapv12 = False
+  WSDL = "https://aws.afip.gov.ar/sr-padron/webservices/personaServiceA5?WSDL"
+  URL = "https://aws.afip.gov.ar/sr-padron/webservices/personaServiceA5"
+  WSDL_TEST = "https://awshomo.afip.gov.ar/sr-padron/webservices/personaServiceA5?WSDL"
+  URL_TEST = "https://awshomo.afip.gov.ar/sr-padron/webservices/personaServiceA5"
+  
+  def __init__(self, afip):
+    super(RegisterInscriptionProof, self).__init__(afip, { "service": "ws_sr_constancia_inscripcion" })
+
+  # Asks to web service for taxpayer details
+  def getTaxpayerDetails(self, identifier: int):
+    # Get token and sign
+    ta = self.getTokenAuthorization()
+
+    # Prepare params
+    params = {
+      "token": ta["token"],
+      "sign": ta["sign"],
+      "cuitRepresentada": self.afip.CUIT,
+      "idPersona": identifier
+    }
+
+    return self.executeRequest("getPersona_v2", params)
+
+  # Asks to web service for taxpayers details
+  def getTaxpayersDetails(self, identifiers: list):
+    # Get token and sign
+    ta = self.getTokenAuthorization()
+
+    # Prepare params
+    params = {
+      "token": ta["token"],
+      "sign": ta["sign"],
+      "cuitRepresentada": self.afip.CUIT,
+      "idPersona": identifiers
+    }
+
+    return self.executeRequest("getPersonaList_v2", params)["persona"]
+
+  # Asks to web service for servers status
+  def getServerStatus(self):
+    return self.executeRequest("dummy")
+
+  # Send request to AFIP servers
+  def executeRequest(self, operation, params = {}):
+    results = super(RegisterInscriptionProof, self).executeRequest(operation, params)
+
+    if operation == "getPersona_v2":
+      return results["personaReturn"]
+    elif operation == "getPersonaList_v2":
+      return results["personaListReturn"]
+    else:
+      return results["return"]

afip/register_scope_ten.py

@@ -0,0 +1,39 @@
+from .web_service import WebService
+
+class RegisterScopeTen(WebService):
+  soapv12 = False
+  WSDL = "https://aws.afip.gov.ar/sr-padron/webservices/personaServiceA10?WSDL"
+  URL = "https://aws.afip.gov.ar/sr-padron/webservices/personaServiceA10"
+  WSDL_TEST = "https://awshomo.afip.gov.ar/sr-padron/webservices/personaServiceA10?WSDL"
+  URL_TEST = "https://awshomo.afip.gov.ar/sr-padron/webservices/personaServiceA10"
+
+  def __init__(self, afip):
+    super(RegisterScopeTen, self).__init__(afip, { "service": "ws_sr_padron_a10" })
+
+  # Asks to web service for taxpayer details
+  def getTaxpayerDetails(self, identifier: int):
+    # Get token and sign
+    ta = self.getTokenAuthorization()
+
+    # Prepare params
+    params = {
+      "token": ta["token"],
+      "sign": ta["sign"],
+      "cuitRepresentada": self.afip.CUIT,
+      "idPersona": identifier
+    }
+
+    return self.executeRequest("getPersona", params)
+
+  # Asks to web service for servers status
+  def getServerStatus(self):
+    return self.executeRequest("dummy")
+
+  # Send request to AFIP servers
+  def executeRequest(self, operation, params = {}):
+    results = super(RegisterScopeTen, self).executeRequest(operation, params)
+
+    if operation == "getPersona":
+      return results["personaReturn"]
+    else:
+      return results["return"]

afip/register_scope_thirteen.py

@@ -0,0 +1,56 @@
+from .web_service import WebService
+
+class RegisterScopeThirteen(WebService):
+  soapv12 = False
+  WSDL = "https://aws.afip.gov.ar/sr-padron/webservices/personaServiceA13?WSDL"
+  URL = "https://aws.afip.gov.ar/sr-padron/webservices/personaServiceA13"
+  WSDL_TEST = "https://awshomo.afip.gov.ar/sr-padron/webservices/personaServiceA13?WSDL"
+  URL_TEST = "https://awshomo.afip.gov.ar/sr-padron/webservices/personaServiceA13"
+
+  def __init__(self, afip):
+    super(RegisterScopeThirteen, self).__init__(afip, { "service": "ws_sr_padron_a13" })
+
+  # Asks to web service for taxpayer details
+  def getTaxpayerDetails(self, identifier: int):
+    # Get token and sign
+    ta = self.getTokenAuthorization()
+
+    # Prepare params
+    params = {
+      "token": ta["token"],
+      "sign": ta["sign"],
+      "cuitRepresentada": self.afip.CUIT,
+      "idPersona": identifier
+    }
+
+    return self.executeRequest("getPersona", params)
+
+  # Asks to web service for tax id by document number
+  def getTaxIDByDocument(self, document_number: int):
+    # Get token and sign
+    ta = self.getTokenAuthorization()
+
+    # Prepare params
+    params = {
+      "token": ta["token"],
+      "sign": ta["sign"],
+      "cuitRepresentada": self.afip.CUIT,
+      "documento": document_number
+    }
+
+    return self.executeRequest("getIdPersonaListByDocumento", params)["idPersona"]
+
+  # Asks to web service for servers status
+  def getServerStatus(self):
+    return self.executeRequest("dummy")
+
+  # Send request to AFIP servers
+  def executeRequest(self, operation, params = {}):
+    results = super(RegisterScopeThirteen, self).executeRequest(operation, params)
+
+    if operation == "getPersona":
+      return results["personaReturn"]
+    elif operation == "getIdPersonaListByDocumento":
+      return results["idPersonaListReturn"]
+    else:
+      return results["return"]

afip/web_service.py

@@ -0,0 +1,55 @@
+import http.client
+import json
+
+class WebService:
+  def __init__(self, afip, options: dict = {}):
+    self.afip = afip
+    self.options = options
+    self.WSDL = options.get('WSDL')
+    self.URL = options.get('URL')
+    self.WSDL_TEST = options.get('WSDL_TEST')
+    self.URL_TEST = options.get('URL_TEST')
+    self.soapv12 = options.get('soapV1_2')
+
+    if not(options.get('service')):
+      raise Exception("service field is required in options")
+    
+  # Gets token authorization for an AFIP Web Service
+  #
+  # If force is true it forces to create a new TA
+  def getTokenAuthorization(self, force = False) -> dict:
+    return self.afip.getServiceTA(self.options.get('service'), force)
+
+  # Sends request to AFIP servers
+  def executeRequest(self, method, params = {}) -> dict:
+    conn = http.client.HTTPSConnection("app.afipsdk.com")
+
+    payload = {
+      "method": method,
+      "params": params,
+      "environment": self.afip.environment,
+      "wsid": self.options.get('service'),
+      "url": self.URL if self.afip.production else self.URL_TEST,
+      "wsdl": self.WSDL if self.afip.production else self.WSDL_TEST,
+      "soap_v_1_2": self.soapv12
+    }
+
+    headers = {
+      "Content-Type": "application/json",
+      "sdk-version-number": self.afip.sdk_version_number,
+      "sdk-library": "python",
+      "sdk-environment": self.afip.environment
+    }
+
+    if self.afip.access_token: headers["Authorization"] = "Bearer %s" % self.afip.access_token
+
+    conn.request("POST", "/api/v1/afip/requests", json.dumps(payload), headers)
+
+    res = conn.getresponse()
+    
+    data = res.read()
+
+    if res.getcode() >= 400:
+      raise Exception(data.decode("utf-8"))
+
+    return json.loads(data.decode("utf-8"))