98 lines
2.2 KiB
Python
98 lines
2.2 KiB
Python
from dataclasses import dataclass
|
|
from urllib.parse import quote_plus, urlparse
|
|
|
|
import requests
|
|
from aws_lambda_powertools.event_handler.exceptions import BadRequestError
|
|
from glom import glom
|
|
from layercake.strutils import random_str
|
|
|
|
from config import KONVIVA_API_URL, KONVIVA_SECRET_KEY
|
|
|
|
ALU = 2
|
|
GES = 3
|
|
|
|
|
|
headers = {
|
|
'Authorization': f'KONVIVA {KONVIVA_SECRET_KEY}',
|
|
'Content-Type': 'application/json',
|
|
}
|
|
|
|
|
|
class KonvivaError(BadRequestError):
|
|
pass
|
|
|
|
|
|
class EmailAlreadyExistsError(KonvivaError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class User:
|
|
IDUsuario: int
|
|
Identificador: str
|
|
NomeUsuario: str
|
|
Email: str
|
|
CPF: str | None = None
|
|
|
|
|
|
def create_user(
|
|
id: str,
|
|
name: str,
|
|
email: str,
|
|
cpf: str | None,
|
|
) -> dict:
|
|
url = urlparse(KONVIVA_API_URL)._replace(
|
|
path='/action/api/integrarUsuario',
|
|
query='sendMail=false',
|
|
)
|
|
|
|
r = requests.post(
|
|
url=url.geturl(),
|
|
headers=headers,
|
|
json={
|
|
'Identificador': id,
|
|
'NomeUsuario': name,
|
|
'Login': email,
|
|
'Email': email,
|
|
'CPF': cpf,
|
|
'Senha': random_str(30),
|
|
'Situacao': 'ATIVO',
|
|
'UnidadesPerfil': [
|
|
{
|
|
'IDPerfil': ALU,
|
|
'CODPerfil': 'ALU',
|
|
'CODUnidade': '1380e6d0-fee8-4cb9-9bf0-a087e703b1a4',
|
|
}
|
|
],
|
|
},
|
|
)
|
|
r.raise_for_status()
|
|
|
|
# Because Konviva does not return the proper HTTP status code
|
|
if err := glom(r.json(), 'errors', default=None):
|
|
err = err[0] if isinstance(err, list) else err
|
|
|
|
if err == 'Login já existente':
|
|
raise EmailAlreadyExistsError(err)
|
|
else:
|
|
raise KonvivaError(err)
|
|
|
|
return r.json().get('IDUsuario')
|
|
|
|
|
|
def get_users_by_email(email: str) -> list[dict]:
|
|
url = urlparse(KONVIVA_API_URL)._replace(
|
|
path='/action/api/getUsuariosByQuery',
|
|
query=f'q=Email=={quote_plus(email)} or Login=={quote_plus(email)}',
|
|
)
|
|
r = requests.get(
|
|
url=url.geturl(),
|
|
headers=headers,
|
|
)
|
|
r.raise_for_status()
|
|
|
|
if glom(r.json(), '0.errors', default=None):
|
|
return []
|
|
|
|
return r.json()
|