notify-api-412 use black to enforce python coding style

This commit is contained in:
Kenneth Kehl
2023-08-25 09:12:23 -07:00
parent c6eb007386
commit 8c9721d8e2
201 changed files with 31660 additions and 28105 deletions

View File

@@ -24,36 +24,35 @@ from app.utils.user_permissions import (
def _get_service_id_from_view_args():
return str(request.view_args.get('service_id', '')) or None
return str(request.view_args.get("service_id", "")) or None
def _get_org_id_from_view_args():
return str(request.view_args.get('org_id', '')) or None
return str(request.view_args.get("org_id", "")) or None
class User(JSONModel, UserMixin):
MAX_FAILED_LOGIN_COUNT = 10
ALLOWED_PROPERTIES = {
'id',
'name',
'email_address',
'auth_type',
'current_session_id',
'failed_login_count',
'email_access_validated_at',
'logged_in_at',
'mobile_number',
'password_changed_at',
'permissions',
'state',
"id",
"name",
"email_address",
"auth_type",
"current_session_id",
"failed_login_count",
"email_access_validated_at",
"logged_in_at",
"mobile_number",
"password_changed_at",
"permissions",
"state",
}
def __init__(self, _dict):
super().__init__(_dict)
self.permissions = _dict.get('permissions', {})
self._platform_admin = _dict['platform_admin']
self.permissions = _dict.get("permissions", {})
self._platform_admin = _dict["platform_admin"]
@classmethod
def from_id(cls, user_id):
@@ -104,8 +103,7 @@ class User(JSONModel, UserMixin):
"""
self._permissions = {
service: translate_permissions_from_db_to_ui(permissions)
for service, permissions
in permissions_by_service.items()
for service, permissions in permissions_by_service.items()
}
def update(self, **kwargs):
@@ -117,9 +115,7 @@ class User(JSONModel, UserMixin):
self.__init__(response)
def update_email_access_validated_at(self):
self.update(
email_access_validated_at=datetime.utcnow().isoformat()
)
self.update(email_access_validated_at=datetime.utcnow().isoformat())
def password_changed_more_recently_than(self, datetime_string):
if not self.password_changed_at:
@@ -144,24 +140,26 @@ class User(JSONModel, UserMixin):
)
def logged_in_elsewhere(self):
return session.get('current_session_id') != self.current_session_id
return session.get("current_session_id") != self.current_session_id
def activate(self):
if self.is_pending:
user_data = user_api_client.activate_user(self.id)
return self.__class__(user_data['data'])
return self.__class__(user_data["data"])
else:
return self
def login(self):
login_user(self)
session['user_id'] = self.id
session["user_id"] = self.id
def send_login_code(self):
if self.email_auth:
user_api_client.send_verify_code(self.id, 'email', None, request.args.get('next'))
user_api_client.send_verify_code(
self.id, "email", None, request.args.get("next")
)
if self.sms_auth:
user_api_client.send_verify_code(self.id, 'sms', self.mobile_number)
user_api_client.send_verify_code(self.id, "sms", self.mobile_number)
def sign_out(self):
session.clear()
@@ -172,22 +170,22 @@ class User(JSONModel, UserMixin):
@property
def sms_auth(self):
return self.auth_type == 'sms_auth'
return self.auth_type == "sms_auth"
@property
def email_auth(self):
return self.auth_type == 'email_auth'
return self.auth_type == "email_auth"
def reset_failed_login_count(self):
user_api_client.reset_failed_login_count(self.id)
@property
def is_active(self):
return self.state == 'active'
return self.state == "active"
@property
def is_pending(self):
return self.state == 'pending'
return self.state == "pending"
@property
def is_gov_user(self):
@@ -197,20 +195,25 @@ class User(JSONModel, UserMixin):
@property
def is_authenticated(self):
return (
not self.logged_in_elsewhere() and
super(User, self).is_authenticated
)
return not self.logged_in_elsewhere() and super(User, self).is_authenticated
@property
def platform_admin(self):
current_app.logger.warn(f"Checking User {self.id} for platform admin: {self._platform_admin}")
return self._platform_admin and not session.get('disable_platform_admin_view', False)
current_app.logger.warn(
f"Checking User {self.id} for platform admin: {self._platform_admin}"
)
return self._platform_admin and not session.get(
"disable_platform_admin_view", False
)
def has_permissions(self, *permissions, restrict_admin_usage=False, allow_org_user=False):
def has_permissions(
self, *permissions, restrict_admin_usage=False, allow_org_user=False
):
unknown_permissions = set(permissions) - all_ui_permissions
if unknown_permissions:
raise TypeError('{} are not valid permissions'.format(list(unknown_permissions)))
raise TypeError(
"{} are not valid permissions".format(list(unknown_permissions))
)
# Service id is always set on the request for service specific views.
service_id = _get_service_id_from_view_args()
@@ -236,9 +239,7 @@ class User(JSONModel, UserMixin):
current_app.logger.warn(f"{log_msg} True because belongs_to_service")
return True
if any(
self.permissions_for_service(service_id) & set(permissions)
):
if any(self.permissions_for_service(service_id) & set(permissions)):
current_app.logger.warn(f"{log_msg} permissions valid")
return True
@@ -266,7 +267,7 @@ class User(JSONModel, UserMixin):
return True
# Top-level templates are always visible
if template_folder is None or template_folder['id'] is None:
if template_folder is None or template_folder["id"] is None:
return True
return self.id in template_folder.get("users_with_permission", [])
@@ -297,7 +298,7 @@ class User(JSONModel, UserMixin):
@property
def email_domain(self):
return self.email_address.split('@')[-1]
return self.email_address.split("@")[-1]
@cached_property
def orgs_and_services(self):
@@ -306,38 +307,36 @@ class User(JSONModel, UserMixin):
@property
def services(self):
from app.models.service import Services
return Services(self.orgs_and_services['services'])
return Services(self.orgs_and_services["services"])
@property
def services_with_organization(self):
return [
service for service in self.services
service
for service in self.services
if self.belongs_to_organization(service.organization_id)
]
@property
def service_ids(self):
return self._dict['services']
return self._dict["services"]
@property
def trial_mode_services(self):
return [
service for service in self.services if service.trial_mode
]
return [service for service in self.services if service.trial_mode]
@property
def live_services(self):
return [
service for service in self.services if service.live
]
return [service for service in self.services if service.live]
@property
def organizations(self):
return Organizations(self.orgs_and_services['organizations'])
return Organizations(self.orgs_and_services["organizations"])
@property
def organization_ids(self):
return self._dict['organizations']
return self._dict["organizations"]
@cached_property
def default_organization(self):
@@ -351,11 +350,7 @@ class User(JSONModel, UserMixin):
@property
def has_access_to_live_and_trial_mode_services(self):
return (
self.organizations or self.live_services
) and (
self.trial_mode_services
)
return (self.organizations or self.live_services) and (self.trial_mode_services)
def serialize(self):
dct = {
@@ -368,10 +363,10 @@ class User(JSONModel, UserMixin):
"failed_login_count": self.failed_login_count,
"permissions": [x for x in self._permissions],
"organizations": self.organization_ids,
"current_session_id": self.current_session_id
"current_session_id": self.current_session_id,
}
if hasattr(self, '_password'):
dct['password'] = self._password
if hasattr(self, "_password"):
dct["password"] = self._password
return dct
@classmethod
@@ -383,13 +378,15 @@ class User(JSONModel, UserMixin):
password,
auth_type,
):
return cls(user_api_client.register_user(
name,
email_address,
mobile_number or None,
password,
auth_type,
))
return cls(
user_api_client.register_user(
name,
email_address,
mobile_number or None,
password,
auth_type,
)
)
def set_password(self, pwd):
self._password = pwd
@@ -398,16 +395,20 @@ class User(JSONModel, UserMixin):
user_api_client.send_verify_email(self.id, self.email_address)
def send_verify_code(self, to=None):
user_api_client.send_verify_code(self.id, 'sms', to or self.mobile_number)
user_api_client.send_verify_code(self.id, "sms", to or self.mobile_number)
def send_already_registered_email(self):
user_api_client.send_already_registered_email(self.id, self.email_address)
def refresh_session_id(self):
self.current_session_id = user_api_client.get_user(self.id).get('current_session_id')
session['current_session_id'] = self.current_session_id
self.current_session_id = user_api_client.get_user(self.id).get(
"current_session_id"
)
session["current_session_id"] = self.current_session_id
def add_to_service(self, service_id, permissions, folder_permissions, invited_by_id):
def add_to_service(
self, service_id, permissions, folder_permissions, invited_by_id
):
try:
user_api_client.add_user_to_service(
service_id,
@@ -422,7 +423,10 @@ class User(JSONModel, UserMixin):
ui_permissions=permissions,
)
except HTTPError as exception:
if exception.status_code == 400 and 'already part of service' in exception.message:
if (
exception.status_code == 400
and "already part of service" in exception.message
):
pass
else:
raise exception
@@ -442,22 +446,21 @@ class User(JSONModel, UserMixin):
class InvitedUser(JSONModel):
ALLOWED_PROPERTIES = {
'id',
'service',
'email_address',
'permissions',
'status',
'created_at',
'auth_type',
'folder_permissions',
"id",
"service",
"email_address",
"permissions",
"status",
"created_at",
"auth_type",
"folder_permissions",
}
def __init__(self, _dict):
super().__init__(_dict)
self.permissions = _dict.get('permissions') or []
self._from_user = _dict['from_user']
self.permissions = _dict.get("permissions") or []
self._from_user = _dict["from_user"]
@classmethod
def create(
@@ -469,14 +472,16 @@ class InvitedUser(JSONModel):
auth_type,
folder_permissions,
):
return cls(invite_api_client.create_invite(
invite_from_id,
service_id,
email_address,
permissions,
auth_type,
folder_permissions,
))
return cls(
invite_api_client.create_invite(
invite_from_id,
service_id,
email_address,
permissions,
auth_type,
folder_permissions,
)
)
@classmethod
def by_id_and_service_id(cls, service_id, invited_user_id):
@@ -486,9 +491,7 @@ class InvitedUser(JSONModel):
@classmethod
def by_id(cls, invited_user_id):
return cls(
invite_api_client.get_invited_user(invited_user_id)
)
return cls(invite_api_client.get_invited_user(invited_user_id))
def accept_invite(self):
invite_api_client.accept_invite(self.service, self.id)
@@ -502,7 +505,7 @@ class InvitedUser(JSONModel):
if isinstance(permissions, list):
self._permissions = permissions
else:
self._permissions = permissions.split(',')
self._permissions = permissions.split(",")
self._permissions = translate_permissions_from_db_to_ui(self.permissions)
@property
@@ -511,110 +514,122 @@ class InvitedUser(JSONModel):
@property
def sms_auth(self):
return self.auth_type == 'sms_auth'
return self.auth_type == "sms_auth"
@property
def email_auth(self):
return self.auth_type == 'email_auth'
return self.auth_type == "email_auth"
@classmethod
def from_token(cls, token):
try:
return cls(invite_api_client.check_token(token))
except HTTPError as exception:
if exception.status_code == 400 and 'invitation' in exception.message:
raise InviteTokenError(exception.message['invitation'])
if exception.status_code == 400 and "invitation" in exception.message:
raise InviteTokenError(exception.message["invitation"])
else:
raise exception
@classmethod
def from_session(cls):
invited_user_id = session.get('invited_user_id')
invited_user_id = session.get("invited_user_id")
return cls.by_id(invited_user_id) if invited_user_id else None
def has_permissions(self, *permissions):
current_app.logger.warn(f"Checking invited user {self.id} for permissions: {permissions}")
if self.status == 'cancelled':
current_app.logger.warn(
f"Checking invited user {self.id} for permissions: {permissions}"
)
if self.status == "cancelled":
return False
return set(self.permissions) > set(permissions)
def has_permission_for_service(self, service_id, permission):
current_app.logger.warn(f"Checking invited user {self.id} for permission: {permission} on service {service_id}")
if self.status == 'cancelled':
current_app.logger.warn(
f"Checking invited user {self.id} for permission: {permission} on service {service_id}"
)
if self.status == "cancelled":
return False
return self.service == service_id and permission in self.permissions
def __eq__(self, other):
return ((self.id,
self.service,
self._from_user,
self.email_address,
self.auth_type,
self.status) == (other.id,
other.service,
other._from_user,
other.email_address,
other.auth_type,
other.status))
return (
self.id,
self.service,
self._from_user,
self.email_address,
self.auth_type,
self.status,
) == (
other.id,
other.service,
other._from_user,
other.email_address,
other.auth_type,
other.status,
)
def serialize(self, permissions_as_string=False):
data = {'id': self.id,
'service': self.service,
'from_user': self._from_user,
'email_address': self.email_address,
'status': self.status,
'created_at': str(self.created_at),
'auth_type': self.auth_type,
'folder_permissions': self.folder_permissions
}
data = {
"id": self.id,
"service": self.service,
"from_user": self._from_user,
"email_address": self.email_address,
"status": self.status,
"created_at": str(self.created_at),
"auth_type": self.auth_type,
"folder_permissions": self.folder_permissions,
}
if permissions_as_string:
data['permissions'] = ','.join(self.permissions)
data["permissions"] = ",".join(self.permissions)
else:
data['permissions'] = sorted(self.permissions)
data["permissions"] = sorted(self.permissions)
return data
def template_folders_for_service(self, service=None):
# only used on the manage users page to display the count, so okay to not be fully fledged for now
return [{'id': x} for x in self.folder_permissions]
return [{"id": x} for x in self.folder_permissions]
def is_editable_by(self, other):
return False
class InvitedOrgUser(JSONModel):
ALLOWED_PROPERTIES = {
'id',
'organization',
'email_address',
'status',
'created_at',
"id",
"organization",
"email_address",
"status",
"created_at",
}
def __init__(self, _dict):
super().__init__(_dict)
self._invited_by = _dict['invited_by']
self._invited_by = _dict["invited_by"]
def __eq__(self, other):
return ((self.id,
self.organization,
self._invited_by,
self.email_address,
self.status) == (other.id,
other.organization,
other._invited_by,
other.email_address,
other.status))
return (
self.id,
self.organization,
self._invited_by,
self.email_address,
self.status,
) == (
other.id,
other.organization,
other._invited_by,
other.email_address,
other.status,
)
@classmethod
def create(cls, invite_from_id, org_id, email_address):
return cls(org_invite_api_client.create_invite(
invite_from_id, org_id, email_address
))
return cls(
org_invite_api_client.create_invite(invite_from_id, org_id, email_address)
)
@classmethod
def from_session(cls):
invited_org_user_id = session.get('invited_org_user_id')
invited_org_user_id = session.get("invited_org_user_id")
return cls.by_id(invited_org_user_id) if invited_org_user_id else None
@classmethod
@@ -625,18 +640,17 @@ class InvitedOrgUser(JSONModel):
@classmethod
def by_id(cls, invited_user_id):
return cls(
org_invite_api_client.get_invited_user(invited_user_id)
)
return cls(org_invite_api_client.get_invited_user(invited_user_id))
def serialize(self, permissions_as_string=False):
data = {'id': self.id,
'organization': self.organization,
'invited_by': self._invited_by,
'email_address': self.email_address,
'status': self.status,
'created_at': str(self.created_at)
}
data = {
"id": self.id,
"organization": self.organization,
"invited_by": self._invited_by,
"email_address": self.email_address,
"status": self.status,
"created_at": str(self.created_at),
}
return data
@property
@@ -648,8 +662,8 @@ class InvitedOrgUser(JSONModel):
try:
return cls(org_invite_api_client.check_token(token))
except HTTPError as exception:
if exception.status_code == 400 and 'invitation' in exception.message:
raise InviteTokenError(exception.message['invitation'])
if exception.status_code == 400 and "invitation" in exception.message:
raise InviteTokenError(exception.message["invitation"])
else:
raise exception
@@ -669,7 +683,6 @@ class AnonymousUser(AnonymousUserMixin):
class Users(ModelList):
client_method = user_api_client.get_users_for_service
model = User
@@ -683,7 +696,7 @@ class Users(ModelList):
user = User.from_id(id)
if user and user.name:
return user.name
return 'Unknown'
return "Unknown"
class OrganizationUsers(Users):
@@ -691,14 +704,14 @@ class OrganizationUsers(Users):
class InvitedUsers(Users):
client_method = invite_api_client.get_invites_for_service
model = InvitedUser
def __init__(self, service_id):
self.items = [
user for user in self.client_method(service_id)
if user['status'] != 'accepted'
user
for user in self.client_method(service_id)
if user["status"] != "accepted"
]