Users¶
Users(baze)
¶
Class used for handling Users.
Source code in echo_baze/baze_root.py
def __init__(self, baze: e_bz.Baze) -> None:
"""Base class that all subclasses should inherit from.
Parameters
----------
baze : Baze
Top level object carrying all functionality and the connection handler.
"""
# check inputs
if not isinstance(baze, e_bz.Baze):
raise ValueError(f"baze must be of type Baze, not {type(baze)}")
self.baze: e_bz.Baze = baze
get(output_type='dict', names=None)
¶
Gets all users with detailed information.
The most useful keys/columns returned are:
- id
- fullName
- invalidLoginAttempts
- lastLoginAttempt
- lockedDate
- roles
Parameters:
-
(output_type¶Literal['dict', 'DataFrame'], default:'dict') –Output type of the data. Can be one of ["dict", "DataFrame"] By default "dict"
-
(names¶list[str] | None, default:None) –List of names of the users to be retrieved. If this is provided, more information will be returned. If None, all users will be retrieved, by default None
Returns:
-
dict[str, dict[str, Any]]–In case output_type == "dict" it will return a dict with the following format: {name: {attribute: value, ...}, ...}
-
DataFrame–In case output_type == "DataFrame" it will return a DataFrame with the following format: index = name, columns = [attribute, ...]
Source code in echo_baze/users.py
@validate_call
def get(
self,
output_type: Literal["dict", "DataFrame"] = "dict",
names: list[str] | None = None,
) -> dict[str, dict[str, Any]] | DataFrame:
"""Gets all users with detailed information.
The most useful keys/columns returned are:
- id
- email
- fullName
- invalidLoginAttempts
- lastLoginAttempt
- lockedDate
- roles
Parameters
----------
output_type : Literal["dict", "DataFrame"], optional
Output type of the data. Can be one of ["dict", "DataFrame"]
By default "dict"
names : list[str] | None, optional
List of names of the users to be retrieved. If this is provided, more information will be returned.
If None, all users will be retrieved, by default None
Returns
-------
dict[str, dict[str, Any]]
In case output_type == "dict" it will return a dict with the following format: {name: {attribute: value, ...}, ...}
DataFrame
In case output_type == "DataFrame" it will return a DataFrame with the following format: index = name, columns = [attribute, ...]
"""
payload = {
"includeRoles": True,
"includePermission": True,
}
# getting all
if names is None:
endpoint = "users"
# getting the data
result = self.baze.conn.get(endpoint, json=payload)
self._handle_http_errors(result)
# converting to dict
result: list[dict[str, Any]] = result.json()
else:
ids = self.get_ids()
result = []
for name in names:
if name not in ids:
logger.warning(f"User with name '{name}' does not exist, skipping")
endpoint = f"users/{ids[name]}"
# getting the data
this_result = self.baze.conn.get(endpoint)
self._handle_http_errors(this_result)
result.append(this_result.json())
if not result:
raise ValueError("None of the desired user names exist!")
# returning on the desired format
match output_type:
case "dict":
# defining the name as the key for the dict
result_dict = {entry["userName"]: entry for entry in result}
# removing the name from the dict values
for entry in result_dict.values():
del entry["userName"]
return result_dict
case "DataFrame":
df = json_normalize(result, max_level=1)
df = df.convert_dtypes(dtype_backend="pyarrow")
df = df.set_index("userName")
return df
case _:
raise ValueError(f"Invalid output_type: {output_type}")
get_identity()
¶
Gets the identity of the current user.
The most useful keys returned are: - id - userName - email - fullName - firstName - lastName - culture - disabled
Returns:
-
dict[str, Any]–Dictionary with the identity of the current user.
Source code in echo_baze/users.py
def get_identity(self) -> dict[str, Any]:
"""Gets the identity of the current user.
The most useful keys returned are:
- id
- userName
- email
- fullName
- firstName
- lastName
- culture
- disabled
Returns
-------
dict[str, Any]
Dictionary with the identity of the current user.
"""
endpoint = "auth/identity"
# getting the data
result = self.baze.conn.get(endpoint)
self._handle_http_errors(result)
# converting to dict
result: dict[str, Any] = result.json()
return result
get_ids()
¶
Gets a dictionary with all user names and ids.
Returns:
-
dict[str, int]:–Dictionary with all users in the format {name: id, ...}
Source code in echo_baze/users.py
def get_ids(self) -> dict[str, int]:
"""Gets a dictionary with all user names and ids.
Returns
-------
dict[str, int]:
Dictionary with all users in the format {name: id, ...}
"""
endpoint = "users"
payload = {
"includeRoles": False,
"includePermission": False,
}
# getting the data
result = self.baze.conn.get(endpoint, json=payload)
self._handle_http_errors(result)
# converting to dict
result: list[dict[str, Any]] = result.json()
return {entry["userName"]: entry["id"] for entry in result}
update(name, new_name=None, full_name=None, culture=None, email=None, customer=None, locked=None, password=None)
¶
Updates a user. Only arguments that are not None will be updated, the rest will be left as they are now.
Parameters:
-
(name¶str) –Name of the user to update.
-
(new_name¶str | None, default:None) –New for the user (will change the one provided), by default None
-
(full_name¶str | None, default:None) –Full name for the user, by default None
-
(culture¶str | None, default:None) –Culture of the user (like pt-BR, en-US, etc.), by default None
-
(email¶str | None, default:None) –Email of the user, by default None
-
(customer¶str | None, default:None) –Customer name, by default None
-
(locked¶bool | None, default:None) –Set to True to lock the user, by default None
-
(password¶str | None, default:None) –New password for the user. Must follow Bazefield password policy. By default None
Source code in echo_baze/users.py
@validate_call
def update(
self,
name: str,
new_name: str | None = None,
full_name: str | None = None,
culture: str | None = None,
email: str | None = None,
customer: str | None = None,
locked: bool | None = None,
password: str | None = None,
) -> None:
"""Updates a user. Only arguments that are not None will be updated, the rest will be left as they are now.
Parameters
----------
name : str
Name of the user to update.
new_name : str | None, optional
New for the user (will change the one provided), by default None
full_name : str | None, optional
Full name for the user, by default None
culture : str | None, optional
Culture of the user (like pt-BR, en-US, etc.), by default None
email : str | None, optional
Email of the user, by default None
customer : str | None, optional
Customer name, by default None
locked : bool | None, optional
Set to True to lock the user, by default None
password : str | None, optional
New password for the user. Must follow Bazefield password policy. By default None
"""
# TODO add the option to disable the user (initial tests didn't seem to work, so we initially removed this option)
args = {
"new_name": {"value": new_name, "type": "name"},
"fullName": {"value": full_name, "type": "normal"},
"culture": {"value": culture, "type": "normal"},
"email": {"value": email, "type": "normal"},
"customer": {"value": customer, "type": "meta"},
"locked": {"value": locked, "type": "locked"},
"password": {"value": password, "type": "password"},
}
# checking if user with desired name exists
if name not in self.get_ids():
raise ValueError(f"User with name '{name}' does not exist")
# checking if at least one of the arguments is not None
if all(arg["value"] is None for arg in args.values()):
raise ValueError("At least one of the arguments must be different from None")
# getting the existing
current = self.get(output_type="dict", names=[name])[name]
# creating dict that will be used to update
new = deepcopy(current)
# adding name to new
new["userName"] = name
# updating new with the new values
if new_name is not None:
new["userName"] = new_name
for arg, arg_value in args.items():
if arg_value["type"] == "normal" and arg_value["value"] is not None:
new[arg] = arg_value["value"]
if arg_value["type"] == "meta" and arg_value["value"] is not None:
new["meta"][arg] = arg_value["value"]
if arg_value["type"] == "locked" and arg_value["value"] is not None:
if arg_value["value"]:
new["meta"]["locked"] = "true"
elif "locked" in new["meta"]:
del new["meta"]["locked"]
if arg_value["type"] == "password" and arg_value["value"] is not None:
new["password"] = arg_value["value"]
# updating modified date using format 0001-01-01T00:00:00.0000000
new["modifiedDate"] = datetime.now(UTC).isoformat()
endpoint = "users"
# sending the data
result = self.baze.conn.post(endpoint, json=new)
self._handle_http_errors(result)