Add API to look up users by provider user ID.

Author: nrsim

firebase_admin/_auth_utils.py

@@ -102,6 +102,15 @@ def validate_provider_id(provider_id, required=True):
             'string.'.format(provider_id))
     return provider_id
 
+def validate_provider_uid(provider_uid, required=True):
+    if provider_uid is None and not required:
+        return None
+    if not isinstance(provider_uid, six.string_types) or not provider_uid:
+        raise ValueError(
+            'Invalid provider uid: "{0}". Provider uid must be a non-empty '
+            'string.'.format(provider_uid))
+    return provider_uid
+
 def validate_photo_url(photo_url, required=False):
     """Parses and validates the given URL string."""
     if photo_url is None and not required:

firebase_admin/_user_mgt.py

@@ -468,6 +468,24 @@ def get_user(self, **kwargs):
         elif 'phone_number' in kwargs:
             key, key_type = kwargs.pop('phone_number'), 'phone number'
             payload = {'phoneNumber' : [_auth_utils.validate_phone(key, required=True)]}
+        elif 'provider_id' in kwargs and 'provider_uid' in kwargs:
+            provider_id = kwargs.pop('provider_id')
+            provider_uid = kwargs.pop('provider_uid')
+            if provider_id == 'phone':
+                key, key_type = provider_uid, 'phone number'
+                payload = {'phoneNumber' : [_auth_utils.validate_phone(key, required=True)]}
+            elif provider_id == 'password':
+                key, key_type = provider_uid, 'email'
+                payload = {'email' : [_auth_utils.validate_email(key, required=True)]}
+            else:
+                key, key_type = {
+                    'provider_id': provider_id, 'provider_uid': provider_uid
+                }, 'provider_user_id'
+                payload = {
+                    'federated_user_id' : [{
+                        'provider_id': _auth_utils.validate_provider_id(key['provider_id']),
+                        'raw_id': _auth_utils.validate_provider_id(key['provider_uid'])}]
+                }
         else:
             raise TypeError('Unsupported keyword arguments: {0}.'.format(kwargs))
 

firebase_admin/auth.py

@@ -72,6 +72,7 @@
     'get_user',
     'get_user_by_email',
     'get_user_by_phone_number',
+    'get_user_by_provider_user_id',
     'import_users',
     'list_users',
     'revoke_refresh_tokens',
@@ -308,6 +309,28 @@ def get_user_by_phone_number(phone_number, app=None):
     response = user_manager.get_user(phone_number=phone_number)
     return UserRecord(response)
 
+def get_user_by_provider_user_id(provider_id, provider_uid, app=None):
+    """Gets the user data corresponding to the specified provider identifier.
+
+    Args:
+        provider_id: Identifier for the given provider, for example,
+            "google.com" for the Google provider.
+        provider_uid: The user identifier with the given provider.
+        app: An App instance (optional).
+
+    Returns:
+        UserRecord: A UserRecord instance.
+
+    Raises:
+        ValueError: If the given provider identifier is None or empty, or if
+        the given provider user identifier is None or empty.
+        UserNotFoundError: If no user exists by the specified identifiers.
+        FirebaseError: If an error occurs while retrieving the user.
+    """
+    user_manager = _get_auth_service(app).user_manager
+    response = user_manager.get_user(
+        provider_id=provider_id, provider_uid=provider_uid)
+    return UserRecord(response)
 
 def list_users(page_token=None, max_results=_user_mgt.MAX_LIST_USERS_RESULTS, app=None):
     """Retrieves a page of user accounts from a Firebase project.

integration/test_auth.py

@@ -177,6 +177,27 @@ def new_user_with_params():
     yield user
     auth.delete_user(user.uid)
 
+@pytest.fixture
+def imported_user_with_params():
+    random_id, email = _random_id()
+    phone = _random_phone()
+    user = auth.ImportUserRecord(
+        uid=random_id,
+        email=email,
+        phone_number=phone,
+        display_name='Random User',
+        photo_url='https://example.com/photo.png',
+        user_metadata=auth.UserMetadata(100, 150),
+        password_hash=b'password', password_salt=b'NaCl', custom_claims={'admin': True},
+        email_verified=True,
+        disabled=False,
+        provider_data=[auth.UserProvider(uid='test', provider_id='google.com')])
+    hash_alg = auth.UserImportHash.scrypt(
+        b'key', rounds=8, memory_cost=14, salt_separator=b'sep')
+    auth.import_users([user], hash_alg=hash_alg)
+    yield user
+    auth.delete_user(user.uid)
+
 @pytest.fixture
 def new_user_list():
     users = [
@@ -200,24 +221,33 @@ def new_user_email_unverified():
     yield user
     auth.delete_user(user.uid)
 
-def test_get_user(new_user_with_params):
-    user = auth.get_user(new_user_with_params.uid)
-    assert user.uid == new_user_with_params.uid
+def test_get_user(imported_user_with_params):
+    user = auth.get_user(imported_user_with_params.uid)
+    assert user.uid == imported_user_with_params.uid
     assert user.display_name == 'Random User'
-    assert user.email == new_user_with_params.email
-    assert user.phone_number == new_user_with_params.phone_number
+    assert user.email == imported_user_with_params.email
+    assert user.phone_number == imported_user_with_params.phone_number
     assert user.photo_url == 'https://example.com/photo.png'
     assert user.email_verified is True
     assert user.disabled is False
+    assert len(user.provider_data) == 3
+    provider_ids = sorted([provider.provider_id for provider in user.provider_data])
+    assert provider_ids == ['google.com', 'password', 'phone']
 
-    user = auth.get_user_by_email(new_user_with_params.email)
-    assert user.uid == new_user_with_params.uid
-    user = auth.get_user_by_phone_number(new_user_with_params.phone_number)
-    assert user.uid == new_user_with_params.uid
+    user = auth.get_user_by_email(imported_user_with_params.email)
+    assert user.uid == imported_user_with_params.uid
 
-    assert len(user.provider_data) == 2
-    provider_ids = sorted([provider.provider_id for provider in user.provider_data])
-    assert provider_ids == ['password', 'phone']
+    user = auth.get_user_by_phone_number(imported_user_with_params.phone_number)
+    assert user.uid == imported_user_with_params.uid
+
+    user = auth.get_user_by_provider_user_id('phone', imported_user_with_params.phone_number)
+    assert user.uid == imported_user_with_params.uid
+
+    user = auth.get_user_by_provider_user_id('password', imported_user_with_params.email)
+    assert user.uid == imported_user_with_params.uid
+
+    user = auth.get_user_by_provider_user_id('google.com', 'test')
+    assert user.uid == imported_user_with_params.uid
 
 def test_list_users(new_user_list):
     err_msg_template = (

tests/test_user_mgt.py

@@ -217,6 +217,19 @@ def test_get_user_by_phone(self, user_mgt_app):
         _instrument_user_manager(user_mgt_app, 200, MOCK_GET_USER_RESPONSE)
         _check_user_record(auth.get_user_by_phone_number('+1234567890', user_mgt_app))
 
+    def test_invalid_get_user_empty_provider_id(self, user_mgt_app):
+        with pytest.raises(ValueError):
+            auth.get_user_by_provider_user_id("", "test_provider_uid", app=user_mgt_app)
+
+    def test_invalid_get_user_empty_provider_uid(self, user_mgt_app):
+        with pytest.raises(ValueError):
+            auth.get_user_by_provider_user_id("google.com", "", app=user_mgt_app)
+
+    def test_get_user_by_provider_user_id(self, user_mgt_app):
+        _instrument_user_manager(user_mgt_app, 200, MOCK_GET_USER_RESPONSE)
+        _check_user_record(
+            auth.get_user_by_provider_user_id('google.com', 'test_google_id', user_mgt_app))
+
     def test_get_user_non_existing(self, user_mgt_app):
         _instrument_user_manager(user_mgt_app, 200, '{"users":[]}')
         with pytest.raises(auth.UserNotFoundError) as excinfo:
@@ -247,6 +260,17 @@ def test_get_user_by_phone_non_existing(self, user_mgt_app):
         assert excinfo.value.http_response is not None
         assert excinfo.value.cause is None
 
+    def test_get_user_by_provider_user_id_non_existing(self, user_mgt_app):
+        _instrument_user_manager(user_mgt_app, 200, '{"users":[]}')
+        with pytest.raises(auth.UserNotFoundError) as excinfo:
+            auth.get_user_by_provider_user_id('google.com', 'test_google_id', user_mgt_app)
+        error_msg = 'No user record found for the provided provider_user_id: %s.' % str(
+            {'provider_id': 'google.com', 'provider_uid': 'test_google_id'})
+        assert excinfo.value.code == exceptions.NOT_FOUND
+        assert str(excinfo.value) == error_msg
+        assert excinfo.value.http_response is not None
+        assert excinfo.value.cause is None
+
     def test_get_user_http_error(self, user_mgt_app):
         _instrument_user_manager(user_mgt_app, 500, '{"error":{"message": "USER_NOT_FOUND"}}')
         with pytest.raises(auth.UserNotFoundError) as excinfo:
@@ -293,6 +317,16 @@ def test_get_user_by_phone_http_error(self, user_mgt_app):
         assert excinfo.value.http_response is not None
         assert excinfo.value.cause is not None
 
+    def test_get_user_by_provider_user_id_http_error(self, user_mgt_app):
+        _instrument_user_manager(user_mgt_app, 500, '{"error":{"message": "USER_NOT_FOUND"}}')
+        with pytest.raises(auth.UserNotFoundError) as excinfo:
+            auth.get_user_by_provider_user_id('google.com', 'test_google_id', user_mgt_app)
+        error_msg = 'No user record found for the given identifier (USER_NOT_FOUND).'
+        assert excinfo.value.code == exceptions.NOT_FOUND
+        assert str(excinfo.value) == error_msg
+        assert excinfo.value.http_response is not None
+        assert excinfo.value.cause is not None
+
 
 class TestCreateUser: