pyAzul.py

import requests


class Products:
    def __init__(self, api, token):
        """
        :param api: Conta Azul API URL

        :param token: Token generated by auth page
        """

        self.API_URL = api
        self.TOKEN = token
        self.HEADERS = {"Authorization": f"Bearer {self.TOKEN}"}
        self.PRODUCT = '/v1/products'
        self.PRODUCT_CAT = '/v1/product-categories'

    def get_product_byId(self, product_id):
        """Get product by id.

        :param product_id: (Required) (string) Product ID.
        """
        # GET
        response = requests.get(self.API_URL + self.PRODUCT + '/' + product_id, headers=self.HEADERS)
        return response.json()

    def get_products(self, name=None, code=None, barcode=None, page=None, size=None):
        """List all products by filters.


        :param name: (Optional) The name of the product.

        :param code: (Optional) The code of the product.

        :param barcode: (Optional) The barcode of the product.

        :param page: (Optional) The page of the list to be returned.

        :param size: (Optional) The quantity of items in the page to be returned.
        """

        # GET
        products = self.PRODUCT
        if name is not None:
            products += f'?name={name}' if products == self.PRODUCT else f'&name={name}'  # List products by name

        if code is not None:
            products += f'?code={code}' if products == self.PRODUCT else f'&code={code}'  # List products by code

        if barcode is not None:
            products += f'?barcode={barcode}' if products == self.PRODUCT else f'&barcode={barcode}'  # List products by barcode

        if page is not None:
            products += f'?page={page}' if products == self.PRODUCT else f'&page={page}'  # List by number of pages

        if size is not None:
            products += f'?size={size}' if products == self.PRODUCT else f'&size={size}'  # List by quantity of items in the page

        response = requests.get(self.API_URL + products, headers=self.HEADERS)
        return response.json()

    def del_product(self, product_id):
        """Delete product by id.

        :param product_id: The id of the product to be deleted.

        """
        # DELETE
        response = requests.delete(self.API_URL + self.PRODUCT + '/' + product_id, headers=self.HEADERS)
        return response.json()

    def new_product(self, product_list):
        """Create new product.

        :param product_list: (dict) See in examples folder.
        """
        # POST
        response = requests.post(self.API_URL + self.PRODUCT, json=product_list, headers=self.HEADERS)
        return response.json()

    def upt_product(self, product_id, product_body):
        """Update Product.

        :param product_id: (Required) (string) The id of the product to be updated.

        :param product_body: (Required) (dict) The product body to be updated.
        """
        # PUT
        response = requests.put(self.API_URL + self.PRODUCT + '/' + product_id, json=product_body, headers=self.HEADERS)
        return response.json()

    def cat_product(self, name=None, page=None, size=None):
        """List product categories.
        :param name:(Optional) The name of the product category        
        :param page: (Optional) The page of the list to be returned
        :param size: (Optional) The quantity of items in the page to be returned"""

        # GET
        cat_product = self.PRODUCT_CAT
        if name is not None:
            cat_product += f'?name={name}' if cat_product == '/v1/product-categories' else f'&name={name}'
        if page is not None:
            cat_product += f'?page={page}' if cat_product == '/v1/product-categories' else f'&page={page}'
        if size is not None:
            cat_product += f'?size={size}' if cat_product == '/v1/product-categories' else f'&size={size}'

        response = requests.get(self.API_URL + cat_product, headers=self.HEADERS)
        return response.json()

    def cat_product_byId(self, product_id):
        """Get products category by id.

        :param product_id: (Required) (string) The id of the product.
        """
        # GET
        response = requests.get(self.API_URL + self.PRODUCT_CAT + '/' + product_id, headers=self.HEADERS)
        return response.json()


class Services:
    def __init__(self, api, token):
        """
               :param api: Conta Azul API URL

               :param token: Token generated by auth page
               """
        self.API_URL = api
        self.TOKEN = token
        self.HEADERS = {"Authorization": f"Bearer {self.TOKEN}"}
        self.SERVICES = '/v1/services'

    def get_service(self, name=None, code=None, page=None, size=None):
        """List all services.

        :param name:(Optional) The name of the service
        :param code: (Optional) The code of the service       
        :param page: (Optional) The page of the list to be returned
        :param size: (Optional) The quantity of items in the page to be returned

        :return: Services (list/dict json).
        """
        # GET
        services = self.SERVICES
        if name is not None:
            services += f'?name={name}' if services == self.SERVICES else f'&name={name}'  # List service by name

        if code is not None:
            services += f'?code={code}' if services == self.SERVICES else f'&code={code}'  # List service by code

        if page is not None:
            services += f'?page={page}' if services == self.SERVICES else f'&page={page}'  # List by number of pages

        if size is not None:
            services += f'?size={size}' if services == self.SERVICES else f'&size={size}'  # List by quantity of items in the page

        response = requests.get(self.API_URL + services, headers=self.HEADERS)
        return response.json()

    def get_service_byId(self, service_id):
        """Get service by id.

        :param service_id: (Required) (string)  The id of the service.
        """
        # GET
        response = requests.get(self.API_URL + self.SERVICES + '/' + service_id, headers=self.HEADERS)
        return response.json()

    def new_service(self, new_service):
        """Create new service.

        :param new_service: (Required) (dict) The service to be created.
        """
        # POST
        response = requests.post(self.API_URL + self.SERVICES, json=new_service, headers=self.HEADERS)
        return response.json()

    def upt_service(self, service_id, service_body):
        """Update Service.

        :param service_id: (Required) (string) The id of the Service to be updated.

        :param service_body: (Required) (dict) The Service body to be updated.
        """
        # PUT
        response = requests.put(self.API_URL + self.SERVICES + '/' + service_id, json=service_body,
                                headers=self.HEADERS)
        return response.json()

    def del_service(self, service_id):
        """Delete service by Id.

        :param service_id: (Required) (string) The id of the service to be deleted.

        """
        # DELETE
        response = requests.delete(self.API_URL + self.SERVICES + '/' + service_id, headers=self.HEADERS)
        return response


class Sales:
    def __init__(self, api, token):
        """
               :param api: Conta Azul API URL

               :param token: Token generated by auth page
               """
        self.API_URL = api
        self.TOKEN = token
        self.HEADERS = {"Authorization": f"Bearer {self.TOKEN}"}
        self.SALES = '/v1/sales'

    def get_sales(self, emission_start=None, emission_end=None, status=None, customer_id=None, page=None, size=None):
        """Get sales by filters.
        default value: Will return all sales.

        :param emission_start: (string) The sale's emission date start filter. Ex: 01/01/2000
        :param emission_end: (string) The sale's emission date end filter. Ex: 01/01/2000 Need Emission start param.
        :param status: (string) The sale's status. PENDING or COMMITTED
        :param customer_id: (string) The sale's customer id.
        :param page: (string) The page of the list to be returned.
        :param size: (string) The quantity of items in the page to be returned.
        """
        # GET
        sales = self.SALES
        if emission_start is not None:
            new_data = emission_start.replace('/', '%2F')
            sales += f'?emission_start={new_data}' if sales == self.SALES else f'&emission_start={new_data}'

        if emission_end is not None:
            new_data = emission_end.replace('/', '%2F')
            sales += f'?emission_end={new_data}' if sales == self.SALES else f'&emission_end={new_data}'

        if status is not None:
            sales += f'?status={status}' if sales == self.SALES else f'&status={status}'

        if customer_id is not None:
            sales += f'?customer_id={customer_id}' if sales == self.SALES else f'&customer_id={customer_id}'

        if page is not None:
            sales += f'?page={page}' if sales == self.SALES else f'&page={page}'

        if size is not None:
            sales += f'?size={size}' if sales == self.SALES else f'&size={size}'

        response = requests.get(self.API_URL + sales, headers=self.HEADERS)
        return response.json()

    def get_sales_byId(self, sale_id):
        """Get Sales by ID

        :param sale_id: (Required) (string)  The sale id.
        """

        # GET
        response = requests.get(self.API_URL + self.SALES + '/' + sale_id, headers=self.HEADERS)
        return response.json()

    def upt_sale(self, sale_id, sale_body):
        """Update sale

        :param sale_id: (Required)(string) The sale id.
        :param sale_body: (Required) (dict) The sale to be updated.
        """
        # PUT
        response = requests.put(self.API_URL + self.SALES + '/' + sale_id, json=sale_body, headers=self.HEADERS)
        return response

    def del_sale(self, sale_id):
        """"  # Delete sale

        :param sale_id: (Required)(string) The sale id.
        """
        # DELETE
        response = requests.delete(self.API_URL + self.SALES + '/' + sale_id, headers=self.HEADERS)
        return response

    def new_sale(self, sale_body):
        """ Create new sale.

        :param sale_body: (Required) (dict) The sale to be created. See in examples folder.

        """
        # POST
        sales = '/v1/sales'
        response = requests.post(self.API_URL + sales, json=sale_body, headers=self.HEADERS)
        return response.json()


class Customer:
    def __init__(self, api, token):
        """
               :param api: Conta Azul API URL

               :param token: Token generated by auth page
               """

        self.API_URL = api
        self.TOKEN = token
        self.HEADERS = {"Authorization": f"Bearer {self.TOKEN}"}
        self.CUSTOMERS = '/v1/customers'

    def new_customer(self, customer_body):
        """Create new customer.

        :param customer_body: (Required) (dict) The customer to be created.
        See in example folder.
        """
        # POST
        response = requests.post(self.API_URL + self.CUSTOMERS, json=customer_body, headers=self.HEADERS)
        return response.json()

    def get_customer(self, search=None, name=None, company_name=None, document=None, page=None, size=None, status=None):
        """List customers by filters

        :param search: When present, the search will be made in customer's company_name, name, email
            and document(CPF / CNPJ) fields, and other text parameters
            (company_name, name and document) will be ignored.

        :param name: The customer's name.

        :param company_name: The customer's company name.

        :param document: The customer's document number.

        :param page: The page of the list to be returned.

        :param size: The quantity of items in the page to be returned.

        :param status: This field accepts ENABLED or DISABLED status, and it was made to match with the search field.
        When present, it will filter only active or inactive customers. When not present, will be filtered for all customers
        """
        # GET
        customers = self.CUSTOMERS
        if search is not None:
            customers += f'?search={search}' if customers == self.CUSTOMERS else f'&search={search}'

        if name is not None:
            customers += f'?name={name}' if customers == self.CUSTOMERS else f'&name={name}'

        if company_name is not None:
            customers += f'?company_name={company_name}' if customers == self.CUSTOMERS else f'&company_name={company_name}'

        if document is not None:
            customers += f'?document={document}' if customers == self.CUSTOMERS else f'&document={document}'

        if page is not None:
            customers += f'?page={page}' if customers == self.CUSTOMERS else f'&page={page}'

        if size is not None:
            customers += f'?size={size}' if customers == self.CUSTOMERS else f'&size={size}'

        if status is not None:
            customers += f'?status={status}' if customers == self.CUSTOMERS else f'&status={status}'

        response = requests.get(self.API_URL + customers, headers=self.HEADERS)
        return response.json()

    def upt_customer(self, customer_id, customer_body):
        """Update Customer.

        :param customer_id: (Required)(string) The customer id.
        :param customer_body: (Required) (dict) The customer to be updated.
               """
        # PUT
        response = requests.put(self.API_URL + self.CUSTOMERS + '/' + customer_id, json=customer_body, headers=self.HEADERS)
        return response

    def get_customer_byId(self, customer_id):
        """Get customer by ID

        :param customer_id: (Required) (string)  The customer id.
        """

        # GET
        response = requests.get(self.API_URL + self.CUSTOMERS + '/' + customer_id, headers=self.HEADERS)
        return response.json()

    def del_customer(self, customer_id):
        """"  # Delete customer

        :param customer_id: (Required)(string) The customer id.
        """
        # DELETE
        response = requests.delete(self.API_URL + self.CUSTOMERS + '/' + customer_id, headers=self.HEADERS)
        return response

    def dis_customer(self, customer_id):
        """Inactivate customer

        :param customer_id: (Required)(string) The customer id.
        """
        disable = f'/v1/customers/inactivate/{customer_id}'
        response = requests.delete(self.API_URL + disable, headers=self.HEADERS)
        return response

    def lst_customer_contact(self, customer_id, page=None, size=None):
        """List customer contacts.

        :param customer_id: (Required)(string) The customer id.

        :param page: The page of the list to be returned.

        :param size: The quantity of items in the page to be returned.
        """
        #GET
        customers = f'/v1/customers/{customer_id}/contacts'

        if page is not None:
            customers += f'?page={page}' if customers == f'/v1/customers/{customer_id}/contacts' else f'&page={page}'

        if size is not None:
            customers += f'?size={size}' if customers == f'/v1/customers/{customer_id}/contacts' else f'&size={size}'

        response = requests.get(self.API_URL + customers, headers=self.HEADERS)
        return response.json()