"""Contains additional metadata associated with a user account."""

def __init__(self, creation_timestamp=None, last_sign_in_timestamp=None,

last_refresh_timestamp=None):

self._creation_timestamp = _auth_utils.validate_timestamp(

creation_timestamp, 'creation_timestamp')

self._last_sign_in_timestamp = _auth_utils.validate_timestamp(

last_sign_in_timestamp, 'last_sign_in_timestamp')

self._last_refresh_timestamp = _auth_utils.validate_timestamp(

last_refresh_timestamp, 'last_refresh_timestamp')

@property

def creation_timestamp(self):

""" Creation timestamp in milliseconds since the epoch.

return self._creation_timestamp

@property

def last_sign_in_timestamp(self):

""" Last sign in timestamp in milliseconds since the epoch.

return self._last_sign_in_timestamp

@property

def last_refresh_timestamp(self):

"""The time at which the user was last active (ID token refreshed).

"""A collection of standard profile information for a user.

@property

def uid(self):

"""Returns the user ID of this user."""

raise NotImplementedError

@property

def display_name(self):

"""Returns the display name of this user."""

raise NotImplementedError

@property

def email(self):

"""Returns the email address associated with this user."""

raise NotImplementedError

@property

def phone_number(self):

"""Returns the phone number associated with this user."""

raise NotImplementedError

@property

def photo_url(self):

"""Returns the photo URL of this user."""

raise NotImplementedError

@property

def provider_id(self):

"""Returns the ID of the identity provider.

"""Contains metadata associated with a Firebase user account."""

def __init__(self, data):

super().__init__()

if not isinstance(data, dict):

raise ValueError(f'Invalid data argument: {data}. Must be a dictionary.')

if not data.get('localId'):

raise ValueError('User ID must not be None or empty.')

self._data = data

@property

def uid(self):

"""Returns the user ID of this user.

return self._data.get('localId')

@property

def display_name(self):

"""Returns the display name of this user.

return self._data.get('displayName')

@property

def email(self):

"""Returns the email address associated with this user.

return self._data.get('email')

@property

def phone_number(self):

"""Returns the phone number associated with this user.

return self._data.get('phoneNumber')

@property

def photo_url(self):

"""Returns the photo URL of this user.

return self._data.get('photoUrl')

@property

def provider_id(self):

"""Returns the provider ID of this user.

return 'firebase'

@property

def email_verified(self):

"""Returns whether the email address of this user has been verified.

return bool(self._data.get('emailVerified'))

@property

def disabled(self):

"""Returns whether this user account is disabled.

return bool(self._data.get('disabled'))

@property

def tokens_valid_after_timestamp(self):

"""Returns the time, in milliseconds since the epoch, before which tokens are invalid.

valid_since = self._data.get('validSince')

if valid_since is not None:

return 1000 * int(valid_since)

return 0

@property

def user_metadata(self):

"""Returns additional metadata associated with this user.

def _int_or_none(key):

if key in self._data:

return int(self._data[key])

return None

last_refresh_at_millis = None

last_refresh_at_rfc3339 = self._data.get('lastRefreshAt', None)

if last_refresh_at_rfc3339:

last_refresh_at_millis = int(_rfc3339.parse_to_epoch(last_refresh_at_rfc3339) * 1000)

return UserMetadata(

_int_or_none('createdAt'), _int_or_none('lastLoginAt'), last_refresh_at_millis)

@property

def provider_data(self):

"""Returns a list of UserInfo instances.

providers = self._data.get('providerUserInfo', [])

return [ProviderUserInfo(entry) for entry in providers]

@property

def custom_claims(self):

"""Returns any custom claims set on this user account.

claims = self._data.get('customAttributes')

if claims:

parsed = json.loads(claims)

if parsed != {}:

return parsed

return None

@property

def tenant_id(self):

"""Returns the tenant ID of this user.

"""Contains metadata associated with a user including password hash and salt."""

@property

def password_hash(self):

"""The user's password hash as a base64-encoded string.

password_hash = self._data.get('passwordHash')

# If the password hash is redacted (probably due to missing permissions) then clear it out,

# similar to how the salt is returned. (Otherwise, it *looks* like a b64-encoded hash is

# present, which is confusing.)

if password_hash == B64_REDACTED:

return None

return password_hash

@property

def password_salt(self):

"""The user's password salt as a base64-encoded string.

"""Represents the result of the ``auth.get_users()`` API."""

def __init__(self, users, not_found):

"""Constructs a `GetUsersResult` object.

self._users = users

self._not_found = not_found

@property

def users(self):

"""Set of `UserRecord` instances, corresponding to the set of users

return self._users

@property

def not_found(self):

"""Set of `UserIdentifier` instances that were requested, but not

"""Represents a page of user records exported from a Firebase project.

def __init__(self, download, page_token, max_results):

self._download = download

self._max_results = max_results

self._current = download(page_token, max_results)

@property

def users(self):

"""A list of ``ExportedUserRecord`` instances available in this page."""

return [ExportedUserRecord(user) for user in self._current.get('users', [])]

@property

def next_page_token(self):

"""Page token string for the next page (empty string indicates no more pages)."""

return self._current.get('nextPageToken', '')

@property

def has_next_page(self):

"""A boolean indicating whether more pages are available."""

return bool(self.next_page_token)

def get_next_page(self):

"""Retrieves the next page of user accounts, if available.

if self.has_next_page:

return ListUsersPage(self._download, self.next_page_token, self._max_results)

return None

def iterate_all(self):

"""Retrieves an iterator for user accounts.

"""Represents the result of the ``auth.delete_users()`` API."""

def __init__(self, result, total):

"""Constructs a `DeleteUsersResult` object.

errors = result.errors

self._success_count = total - len(errors)

self._failure_count = len(errors)

self._errors = errors

@property

def success_count(self):

"""Returns the number of users that were deleted successfully (possibly

return self._success_count

@property

def failure_count(self):

"""Returns the number of users that failed to be deleted (possibly

return self._failure_count

@property

def errors(self):

"""A list of `auth.ErrorInfo` instances describing the errors that

"""Represents the results of a `delete_users()` call."""

def __init__(self, errors=None):

"""Constructs a `BatchDeleteAccountsResponse` instance, corresponding to

"""Contains metadata regarding how a user is known by a particular identity provider."""

def __init__(self, data):

super().__init__()

if not isinstance(data, dict):

raise ValueError(f'Invalid data argument: {data}. Must be a dictionary.')

if not data.get('rawId'):

raise ValueError('User ID must not be None or empty.')

self._data = data

@property

def uid(self):

return self._data.get('rawId')

@property

def display_name(self):

return self._data.get('displayName')

@property

def email(self):

return self._data.get('email')

@property

def phone_number(self):

return self._data.get('phoneNumber')

@property

def photo_url(self):

return self._data.get('photoUrl')

@property

def provider_id(self):

"""Contains required continue/state URL with optional Android and iOS settings.

def __init__(

self,

url: str,

handle_code_in_app: Optional[bool] = None,

dynamic_link_domain: Optional[str] = None,

ios_bundle_id: Optional[str] = None,

android_package_name: Optional[str] = None,

android_install_app: Optional[str] = None,

android_minimum_version: Optional[str] = None,

link_domain: Optional[str] = None,

):

if dynamic_link_domain is not None:

warnings.warn(

'dynamic_link_domain is deprecated, use link_domain instead',

DeprecationWarning

)

self.url = url

self.handle_code_in_app = handle_code_in_app

self.dynamic_link_domain = dynamic_link_domain

self.ios_bundle_id = ios_bundle_id

self.android_package_name = android_package_name

self.android_install_app = android_install_app

self.android_minimum_version = android_minimum_version

""" Validates the provided action code settings for email link generation and

parameters = {}

# url

if not settings.url:

raise ValueError("Dynamic action links url is mandatory")

try:

parsed = parse.urlparse(settings.url)

if not parsed.netloc:

raise ValueError(f'Malformed dynamic action links url: "{settings.url}".')

parameters['continueUrl'] = settings.url

except Exception as err:

raise ValueError(f'Malformed dynamic action links url: "{settings.url}".') from err

# handle_code_in_app

if settings.handle_code_in_app is not None:

if not isinstance(settings.handle_code_in_app, bool):

raise ValueError(

f'Invalid value provided for handle_code_in_app: {settings.handle_code_in_app}')

parameters['canHandleCodeInApp'] = settings.handle_code_in_app

# dynamic_link_domain

if settings.dynamic_link_domain is not None:

if not isinstance(settings.dynamic_link_domain, str):

raise ValueError(

f'Invalid value provided for dynamic_link_domain: {settings.dynamic_link_domain}')

parameters['dynamicLinkDomain'] = settings.dynamic_link_domain

# link_domain

if settings.link_domain is not None:

if not isinstance(settings.link_domain, str):

raise ValueError(

f'Invalid value provided for link_domain: {settings.link_domain}')

parameters['linkDomain'] = settings.link_domain

# ios_bundle_id

if settings.ios_bundle_id is not None:

if not isinstance(settings.ios_bundle_id, str):

raise ValueError(

f'Invalid value provided for ios_bundle_id: {settings.ios_bundle_id}')

parameters['iOSBundleId'] = settings.ios_bundle_id

# android_* attributes

if (settings.android_minimum_version or settings.android_install_app) \

and not settings.android_package_name:

raise ValueError("Android package name is required when specifying other Android settings")

if settings.android_package_name is not None:

if not isinstance(settings.android_package_name, str):

raise ValueError(

f'Invalid value provided for android_package_name: {settings.android_package_name}')

parameters['androidPackageName'] = settings.android_package_name

if settings.android_minimum_version is not None:

if not isinstance(settings.android_minimum_version, str):

raise ValueError(

'Invalid value provided for android_minimum_version: '

f'{settings.android_minimum_version}')

parameters['androidMinimumVersion'] = settings.android_minimum_version

if settings.android_install_app is not None:

if not isinstance(settings.android_install_app, bool):

raise ValueError(

f'Invalid value provided for android_install_app: {settings.android_install_app}')

parameters['androidInstallApp'] = settings.android_install_app

"""Provides methods for interacting with the Google Identity Toolkit."""

ID_TOOLKIT_URL = 'https://identitytoolkit.googleapis.com/v1'

def __init__(self, http_client, project_id, tenant_id=None, url_override=None):

self.http_client = http_client

url_prefix = url_override or self.ID_TOOLKIT_URL

self.base_url = f'{url_prefix}/projects/{project_id}'

if tenant_id:

self.base_url += f'/tenants/{tenant_id}'

def get_user(self, **kwargs):

"""Gets the user data corresponding to the provided key."""

if 'uid' in kwargs:

key, key_type = kwargs.pop('uid'), 'user ID'

payload = {'localId' : [_auth_utils.validate_uid(key, required=True)]}

elif 'email' in kwargs:

key, key_type = kwargs.pop('email'), 'email'

payload = {'email' : [_auth_utils.validate_email(key, required=True)]}

elif 'phone_number' in kwargs:

key, key_type = kwargs.pop('phone_number'), 'phone number'

payload = {'phoneNumber' : [_auth_utils.validate_phone(key, required=True)]}

else:

raise TypeError(f'Unsupported keyword arguments: {kwargs}.')

body, http_resp = self._make_request('post', '/accounts:lookup', json=payload)

if not body or not body.get('users'):

raise _auth_utils.UserNotFoundError(

f'No user record found for the provided {key_type}: {key}.',

http_response=http_resp)

return body['users'][0]

def get_users(self, identifiers):

"""Looks up multiple users by their identifiers (uid, email, etc.)

if not identifiers:

return []

if len(identifiers) > 100:

raise ValueError('`identifiers` parameter must have <= 100 entries.')

payload = defaultdict(list)

for identifier in identifiers:

if isinstance(identifier, _user_identifier.UidIdentifier):

payload['localId'].append(identifier.uid)

elif isinstance(identifier, _user_identifier.EmailIdentifier):

payload['email'].append(identifier.email)

elif isinstance(identifier, _user_identifier.PhoneIdentifier):

payload['phoneNumber'].append(identifier.phone_number)

elif isinstance(identifier, _user_identifier.ProviderIdentifier):

payload['federatedUserId'].append({

'providerId': identifier.provider_id,

'rawId': identifier.provider_uid

})

else:

raise ValueError(

f'Invalid entry in "identifiers" list. Unsupported type: {type(identifier)}')

body, http_resp = self._make_request(

'post', '/accounts:lookup', json=payload)

if not http_resp.ok:

raise _auth_utils.UnexpectedResponseError(

'Failed to get users.', http_response=http_resp)

return body.get('users', [])

def list_users(self, page_token=None, max_results=MAX_LIST_USERS_RESULTS):

"""Retrieves a batch of users."""

if page_token is not None:

if not isinstance(page_token, str) or not page_token:

raise ValueError('Page token must be a non-empty string.')

if not isinstance(max_results, int):

raise ValueError('Max results must be an integer.')

if max_results < 1 or max_results > MAX_LIST_USERS_RESULTS:

raise ValueError(

f'Max results must be a positive integer less than {MAX_LIST_USERS_RESULTS}.')

payload = {'maxResults': max_results}

if page_token:

payload['nextPageToken'] = page_token

body, _ = self._make_request('get', '/accounts:batchGet', params=payload)

return body

def create_user(self, uid=None, display_name=None, email=None, phone_number=None,

photo_url=None, password=None, disabled=None, email_verified=None):

"""Creates a new user account with the specified properties."""

payload = {

'localId': _auth_utils.validate_uid(uid),

'displayName': _auth_utils.validate_display_name(display_name),

'email': _auth_utils.validate_email(email),

'phoneNumber': _auth_utils.validate_phone(phone_number),

'photoUrl': _auth_utils.validate_photo_url(photo_url),

'password': _auth_utils.validate_password(password),

'emailVerified': bool(email_verified) if email_verified is not None else None,

'disabled': bool(disabled) if disabled is not None else None,

}

payload = {k: v for k, v in payload.items() if v is not None}

body, http_resp = self._make_request('post', '/accounts', json=payload)

if not body or not body.get('localId'):

raise _auth_utils.UnexpectedResponseError(

'Failed to create new user.', http_response=http_resp)

return body.get('localId')

def update_user(self, uid, display_name=None, email=None, phone_number=None,

photo_url=None, password=None, disabled=None, email_verified=None,

valid_since=None, custom_claims=None, providers_to_delete=None):

"""Updates an existing user account with the specified properties"""

payload = {

'localId': _auth_utils.validate_uid(uid, required=True),

'email': _auth_utils.validate_email(email),

'password': _auth_utils.validate_password(password),

'validSince': _auth_utils.validate_timestamp(valid_since, 'valid_since'),

'emailVerified': bool(email_verified) if email_verified is not None else None,

'disableUser': bool(disabled) if disabled is not None else None,

}

remove = []

remove_provider = _auth_utils.validate_provider_ids(providers_to_delete)

if display_name is not None:

if display_name is DELETE_ATTRIBUTE:

remove.append('DISPLAY_NAME')

else:

payload['displayName'] = _auth_utils.validate_display_name(display_name)

if photo_url is not None:

if photo_url is DELETE_ATTRIBUTE:

remove.append('PHOTO_URL')

else:

payload['photoUrl'] = _auth_utils.validate_photo_url(photo_url)

if remove:

payload['deleteAttribute'] = remove

if phone_number is not None:

if phone_number is DELETE_ATTRIBUTE:

remove_provider.append('phone')

else:

payload['phoneNumber'] = _auth_utils.validate_phone(phone_number)

if custom_claims is not None:

if custom_claims is DELETE_ATTRIBUTE:

custom_claims = {}

json_claims = json.dumps(custom_claims) if isinstance(

custom_claims, dict) else custom_claims

payload['customAttributes'] = _auth_utils.validate_custom_claims(json_claims)

if remove_provider:

payload['deleteProvider'] = list(set(remove_provider))

payload = {k: v for k, v in payload.items() if v is not None}

body, http_resp = self._make_request('post', '/accounts:update', json=payload)

if not body or not body.get('localId'):

raise _auth_utils.UnexpectedResponseError(

f'Failed to update user: {uid}.', http_response=http_resp)

return body.get('localId')

def delete_user(self, uid):

"""Deletes the user identified by the specified user ID."""

_auth_utils.validate_uid(uid, required=True)

body, http_resp = self._make_request('post', '/accounts:delete', json={'localId' : uid})

if not body or not body.get('kind'):

raise _auth_utils.UnexpectedResponseError(

f'Failed to delete user: {uid}.', http_response=http_resp)

def delete_users(self, uids, force_delete=False):

"""Deletes the users identified by the specified user ids.

if not uids:

return BatchDeleteAccountsResponse()

if len(uids) > 1000:

raise ValueError("`uids` paramter must have <= 1000 entries.")

for uid in uids:

_auth_utils.validate_uid(uid, required=True)

body, http_resp = self._make_request('post', '/accounts:batchDelete',

json={'localIds': uids, 'force': force_delete})

if not isinstance(body, dict):

raise _auth_utils.UnexpectedResponseError(

'Unexpected response from server while attempting to delete users.',

http_response=http_resp)

return BatchDeleteAccountsResponse(body.get('errors', []))

def import_users(self, users, hash_alg=None):

"""Imports the given list of users to Firebase Auth."""

try:

if not users or len(users) > MAX_IMPORT_USERS_SIZE:

raise ValueError(

'Users must be a non-empty list with no more than '

f'{MAX_IMPORT_USERS_SIZE} elements.')

if any(not isinstance(u, _user_import.ImportUserRecord) for u in users):

raise ValueError('One or more user objects are invalid.')

except TypeError as err:

raise ValueError('users must be iterable') from err

payload = {'users': [u.to_dict() for u in users]}

if any('passwordHash' in u for u in payload['users']):

if not isinstance(hash_alg, _user_import.UserImportHash):

raise ValueError('A UserImportHash is required to import users with passwords.')

payload.update(hash_alg.to_dict())

body, http_resp = self._make_request('post', '/accounts:batchCreate', json=payload)

if not isinstance(body, dict):

raise _auth_utils.UnexpectedResponseError(

'Failed to import users.', http_response=http_resp)

return body

def generate_email_action_link(self, action_type, email, action_code_settings=None):

"""Fetches the email action links for types

payload = {

'requestType': _auth_utils.validate_action_type(action_type),

'email': _auth_utils.validate_email(email),

'returnOobLink': True

}

if action_code_settings:

payload.update(encode_action_code_settings(action_code_settings))

body, http_resp = self._make_request('post', '/accounts:sendOobCode', json=payload)

if not body or not body.get('oobLink'):

raise _auth_utils.UnexpectedResponseError(

'Failed to generate email action link.', http_response=http_resp)

return body.get('oobLink')

def _make_request(self, method, path, **kwargs):

url = f'{self.base_url}{path}'

try:

return self.http_client.body_and_response(method, url, **kwargs)

except requests.exceptions.RequestException as error:

@property

def items(self):