"""Mambu utilites module.
.. autosummary::
:nosignatures:
:toctree: _autosummary
Exceptions, some API return codes, utility functions, a lot of urlfuncs
(see MambuStruct.__init__ and .connect pydocs for more info on this)
.. warning::
Secutiry WARNING: Imports the configurations from mambuconfig. It surely
is a bad idea. Got to improve this!
.. todo:: status API V2: fullDetails are not compatible
.. todo:: status API V2: pagination managements needs review
.. todo:: status API V2: almost al GET operations seems
compatible. PATCH/POST/DELETE operations needs review
.. todo:: status API V2: testing of EVERYTHING is required """
import logging
import logging.config as logging_config
import os
from codecs import open as copen
import yaml
from .mambuconfig import apipagination, apipwd, apiurl, apiuser, loggingconf, activate_request_session_objects
from .mambugeturl import getmambuurl
import json
import sys
from datetime import datetime, timezone
API_RETURN_CODES = {
"SUCCESS": 0,
"INVALID_PARAMETERS": 4,
"INVALID_LOAN_ACCOUNT_ID": 100,
"INVALID_ACCOUNT_STATE": 105,
"EXCESS_REPAYMENT_ERROR": 110,
"INVALID_REPAYMENT_DATE_ERROR": 111,
"INVALID_STATE_TRANSITION": 116,
"INVALID_ACCOUNT_STATE_FOR_REPAYMENTS": 128,
"INCONSISTENT_SCHEDULE_PRINCIPAL_DUE_WITH_LOAN_AMOUNT": 132,
}
"""
.. deprecated:: 0.8
It's probably not useful. Besides, :any:`mambustruct.MambuStruct.connect()`
method returns the code when an error occurs by default.
"""
OUT_OF_BOUNDS_PAGINATION_LIMIT_VALUE = int(apipagination)
"""Current maximum number of response elements Mambu returns"""
PAGINATIONDETAILS = ["ON", "OFF"]
"""paginationDetails options"""
DETAILSLEVEL = ["BASIC", "FULL"]
"""detailsLevel options"""
MAX_UPLOAD_SIZE = 52428800 # 50Mb
"""upload files maximum size"""
UPLOAD_FILENAME_INVALID_CHARS = "/><|:&?*[]#\*`"
"""invalid characters for an uploaded filename"""
ALLOWED_UPLOAD_MIMETYPES = [
"JPEG",
"PNG",
"GIF",
"BMP",
"TIFF",
"PDF",
"XML",
"DOC",
"DOCX",
"DOCM",
"DOT",
"DOTX",
"DOTM",
"XLS",
"XLSX",
"XLSB",
"PPT",
"PPTX",
"ODT",
"OTT",
"FODT",
"PDF",
"XML",
"TXT",
"CSV",
"PROPERTIES",
"MSG",
"TIF",
"ZIP",
"RTF",
"XLSM",
"ODS",
"ODP",
"EML",
"EMLX",
"HTML",
"MHT",
"MHTML",
"XPS",
"NUMBERS",
"KEY",
"PAGES",
"YAML",
"JSON",
"JASPER",
"JRXML",
]
"""allowed file types for uploads"""
SEARCH_OPERATORS = [
"EQUALS",
"EQUALS_CASE_SENSITIVE",
"DIFFERENT_THAN",
"MORE_THAN",
"LESS_THAN",
"BETWEEN",
"ON",
"AFTER",
"BEFORE",
"BEFORE_INCLUSIVE",
"STARTS_WITH",
"STARTS_WITH_CASE_SENSITIVE",
"IN",
"TODAY",
"THIS_WEEK",
"THIS_MONTH",
"THIS_YEAR",
"LAST_DAYS",
"EMTPY",
"NOT_EMPTY",
]
"""search operators"""
[docs]class MambuPyError(Exception):
"""Default exception"""
[docs]class MambuError(MambuPyError):
"""Default exception from Mambu"""
[docs]class MambuCommError(MambuError):
"""Thrown when communication issues with Mambu arise"""
### More utility functions follow ###
[docs]def strip_consecutive_repeated_char(s, ch):
"""Strip characters in a string which are consecutively repeated.
Useful when in notes or some other free text fields on Mambu, users
capture anything and a lot of capture errors not always detected by
Mambu get through. You want some cleaning? this may be useful.
This is a string processing function.
"""
sdest = ""
for i, c in enumerate(s):
if i != 0 and s[i] == ch and s[i] == s[i - 1]:
continue
sdest += s[i]
return sdest
from urllib import parse as urlparse
[docs]def iri_to_uri(iri):
"""Change an IRI (internationalized R) to an URI.
Used at MambuStruct.connect() method for any requests done to Mambu.
Perfect example of unicode getting in the way.
Illustrative (I hope) example: I have Mambu usernames with special
chars in them. When retrieving them and then trying to build a
MambuUser object with them, I get a BIG problem because of the
unicode chars there. Using this I solved the problem.
"""
parts = urlparse.urlparse(iri)
uri = [part.decode("utf8") for parti, part in enumerate(parts.encode("utf8"))]
return urlparse.urlunparse(uri)
[docs]def encoded_dict(in_dict):
"""Encode every value of a dict to UTF-8.
Useful for POSTing requests on the 'data' parameter of urlencode.
"""
out_dict = {}
for k, v in in_dict.items():
out_dict[k] = v
return out_dict
from time import sleep
import requests
[docs]def _backup_db_previous_prep(callback, logger, kwargs):
list_ret = []
try:
retries = kwargs["retries"]
except KeyError:
retries = -1
list_ret.append(retries)
try:
justbackup = kwargs["justbackup"]
except KeyError:
justbackup = False
list_ret.append(justbackup)
try:
force_download_latest = bool(kwargs["force_download_latest"])
except KeyError:
force_download_latest = False
list_ret.append(force_download_latest)
try:
headers = kwargs["headers"]
except KeyError:
headers = {
"Accept": "application/vnd.mambu.v2+zip",
}
list_ret.append(headers)
logger.info("Mambu DB Backup")
user = kwargs.pop("user", apiuser)
list_ret.append(user)
pwd = kwargs.pop("pwd", apipwd)
list_ret.append(pwd)
data = {"callback": callback}
list_ret.append(data)
return list_ret
[docs]def _backup_db_request(justbackup, data, user, pwd, logger=None):
try:
if not justbackup:
posturl = iri_to_uri(getmambuurl() + "database/backup")
logger.info("open url: " + posturl)
logger.info("data: " + str(data))
resp = requests.post(
posturl,
data=json.dumps(data),
headers={
"content-type": "application/json",
"Accept": "application/vnd.mambu.v2+json",
},
auth=(user, pwd),
)
logger.info(str(resp.content))
logger.info(resp.request.url)
logger.info(resp.request.body)
logger.info(str([(k, v) for k, v in resp.request.headers.items() if k != "Authorization"]))
except Exception as ex:
mess = "Error requesting backup: %s" % repr(ex)
logger.exception(mess)
raise MambuError(mess)
if not justbackup and resp.status_code != 202:
mess = "Error posting request for backup: %s" % resp.content
logger.error(mess)
raise MambuCommError(mess)
[docs]def _backup_db_timeout_mechanism(
justbackup, retries, bool_func, force_download_latest, logger=None
):
value_to_latest = True
while not justbackup and retries and not bool_func():
logger.debug("waiting...")
sleep(10)
retries -= 1
if retries < 0:
retries = -1
if not justbackup and not retries:
mess = "Tired of waiting, giving up..."
logger.warning(mess)
if not force_download_latest:
raise MambuError(mess)
else:
value_to_latest = False
sleep(30)
return value_to_latest
[docs]def _backup_db_request_download_backup(user, pwd, headers, logger=None):
geturl = iri_to_uri(getmambuurl() + "database/backup/LATEST")
logger.info("open url: " + geturl)
resp = requests.get(geturl, auth=(user, pwd), headers=headers)
if resp.status_code != 200:
mess = "Error getting database backup: %s" % resp.content
logger.error(mess)
raise MambuCommError(mess)
return resp
[docs]def _backup_db_post_processing(resp, output_fname, logger=None):
logger.info("saving...")
with open(output_fname, "wb") as fw:
fw.write(resp.content)
[docs]def backup_db(callback, bool_func, output_fname, *args, **kwargs):
"""Backup Mambu Database via REST API.
Makes two calls to Mambu API:
- a POST to request a backup to be made
- a GET, once the backup is ready, to download the latest backup
* callback is a string to a callback URL Mambu will internally call
when the backup is ready to download. You should have a webservice
there to warn you when the backup is ready.
* bool_func is a function you use against your own code to test if the
said backup is ready. This function backup_db manages both the logic
of the request of a backup and the downloading of it too, so
bool_func allows you to have some way on your side to know when this
function will download the backup.
The thing is you have to build a webservice (for the callback)
making some kind of flag turn that your bool_func will read and know
when to say True, telling backup_db to begin the download of the
backup.
* output_fname the name of the file that will hold the downloaded
backup. PLEASE MIND that Mambu sends a ZIP file here.
* user, pwd and url allow you to change the Mambu permissions for
the getmambuurl internally called here.
* retries number of retries for bool_func or -1 for keep waiting.
* just_backup bool if True, skip asking for backup, just download LATEST
* force_download_latest boolean, True to force download even if no
callback is called. False to throw error if callback isn't received
after retries.
* returns a dictionary with info about the download
-latest boolean flag, if the db downloaded was the latest or not
.. todo:: status API V2: compatible
"""
logger = setup_logging("mambupy.backup_db")
# previous preparation
(
retries,
justbackup,
force_download_latest,
headers,
user,
pwd,
data,
) = _backup_db_previous_prep(callback, logger, kwargs)
# POST to request Mambu to prepare backup of its own DB
_backup_db_request(justbackup, data, user, pwd, logger)
# wait & timeout mechanism
data["latest"] = _backup_db_timeout_mechanism(
justbackup, retries, bool_func, force_download_latest, logger
)
# GET request to download LATEST Mambu's DB backup
resp = _backup_db_request_download_backup(user, pwd, headers, logger)
# post-processing
_backup_db_post_processing(resp, output_fname, logger)
# no refactor
logger.info("DONE!")
return data
[docs]def setup_logging(
loggername,
default_level=logging.INFO,
) -> logging.Logger:
""" Get a logger configured with a yaml file. """
if os.path.exists(loggingconf):
with copen(loggingconf, "r", "utf-8") as f:
config = yaml.full_load(f.read())
logging_config.dictConfig(config)
loggr = logging.getLogger(loggername)
else:
loggr = logging.getLogger(loggername)
loggr.setLevel(default_level)
loggr.propagate = True
return loggr
logger = setup_logging("mambupy")
"""Default logger for mambupy module"""