<s> """<STR_LIT>""" <EOL> from . utils import get_identity_attributes , string_types <EOL> class Datastore ( object ) : <EOL> def __init__ ( self , db ) : <EOL> self . db = db <EOL> def commit ( self ) : <EOL> pass <EOL> def put ( self , model ) : <EOL> raise NotImplementedError <EOL> def delete ( self , model ) : <EOL> raise NotImplementedError <EOL> class SQLAlchemyDatastore ( Datastore ) : <EOL> def commit ( self ) : <EOL> self . db . session . commit ( ) <EOL> def put ( self , model ) : <EOL> self . db . session . add ( model ) <EOL> return model <EOL> def delete ( self , model ) : <EOL> self . db . session . delete ( model ) <EOL> class MongoEngineDatastore ( Datastore ) : <EOL> def put ( self , model ) : <EOL> model . save ( ) <EOL> return model <EOL> def delete ( self , model ) : <EOL> model . delete ( ) <EOL> class PeeweeDatastore ( Datastore ) : <EOL> def put ( self , model ) : <EOL> model . save ( ) <EOL> return model <EOL> def delete ( self , model ) : <EOL> model . delete_instance ( ) <EOL> class UserDatastore ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , user_model , role_model ) : <EOL> self . user_model = user_model <EOL> self . role_model = role_model <EOL> def _prepare_role_modify_args ( self , user , role ) : <EOL> if isinstance ( user , string_types ) : <EOL> user = self . find_user ( email = user ) <EOL> if isinstance ( role , string_types ) : <EOL> role = self . find_role ( role ) <EOL> return user , role <EOL> def _prepare_create_user_args ( self , ** kwargs ) : <EOL> kwargs . setdefault ( '<STR_LIT>' , True ) <EOL> roles = kwargs . get ( '<STR_LIT>' , [ ] ) <EOL> for i , role in enumerate ( roles ) : <EOL> rn = role . name if isinstance ( role , self . role_model ) else role <EOL> roles [ i ] = self . find_role ( rn ) <EOL> kwargs [ '<STR_LIT>' ] = roles <EOL> return kwargs <EOL> def get_user ( self , id_or_email ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def find_user ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def find_role ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def add_role_to_user ( self , user , role ) : <EOL> """<STR_LIT>""" <EOL> user , role = self . _prepare_role_modify_args ( user , role ) <EOL> if role not in user . roles : <EOL> user . roles . append ( role ) <EOL> self . put ( user ) <EOL> return True <EOL> return False <EOL> def remove_role_from_user ( self , user , role ) : <EOL> """<STR_LIT>""" <EOL> rv = False <EOL> user , role = self . _prepare_role_modify_args ( user , role ) <EOL> if role in user . roles : <EOL> rv = True <EOL> user . roles . remove ( role ) <EOL> self . put ( user ) <EOL> return rv <EOL> def toggle_active ( self , user ) : <EOL> """<STR_LIT>""" <EOL> user . active = not user . active <EOL> return True <EOL> def deactivate_user ( self , user ) : <EOL> """<STR_LIT>""" <EOL> if user . active : <EOL> user . active = False <EOL> return True <EOL> return False <EOL> def activate_user ( self , user ) : <EOL> """<STR_LIT>""" <EOL> if not user . active : <EOL> user . active = True <EOL> return True <EOL> return False <EOL> def create_role ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> role = self . role_model ( ** kwargs ) <EOL> return self . put ( role ) <EOL> def find_or_create_role ( self , name , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs [ "<STR_LIT:name>" ] = name <EOL> return self . find_role ( name ) or self . create_role ( ** kwargs ) <EOL> def create_user ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs = self . _prepare_create_user_args ( ** kwargs ) <EOL> user = self . user_model ( ** kwargs ) <EOL> return self . put ( user ) <EOL> def delete_user ( self , user ) : <EOL> """<STR_LIT>""" <EOL> self . delete ( user ) <EOL> class SQLAlchemyUserDatastore ( SQLAlchemyDatastore , UserDatastore ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , db , user_model , role_model ) : <EOL> SQLAlchemyDatastore . __init__ ( self , db ) <EOL> UserDatastore . __init__ ( self , user_model , role_model ) <EOL> def get_user ( self , identifier ) : <EOL> if self . _is_numeric ( identifier ) : <EOL> return self . user_model . query . get ( identifier ) <EOL> for attr in get_identity_attributes ( ) : <EOL> query = getattr ( self . user_model , attr ) . ilike ( identifier ) <EOL> rv = self . user_model . query . filter ( query ) . first ( ) <EOL> if rv is not None : <EOL> return rv <EOL> def _is_numeric ( self , value ) : <EOL> try : <EOL> int ( value ) <EOL> except ( TypeError , ValueError ) : <EOL> return False <EOL> return True <EOL> def find_user ( self , ** kwargs ) : <EOL> return self . user_model . query . filter_by ( ** kwargs ) . first ( ) <EOL> def find_role ( self , role ) : <EOL> return self . role_model . query . filter_by ( name = role ) . first ( ) <EOL> class MongoEngineUserDatastore ( MongoEngineDatastore , UserDatastore ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , db , user_model , role_model ) : <EOL> MongoEngineDatastore . __init__ ( self , db ) <EOL> UserDatastore . __init__ ( self , user_model , role_model ) <EOL> def get_user ( self , identifier ) : <EOL> from mongoengine import ValidationError <EOL> try : <EOL> return self . user_model . objects ( id = identifier ) . first ( ) <EOL> except ValidationError : <EOL> pass <EOL> for attr in get_identity_attributes ( ) : <EOL> query_key = '<STR_LIT>' % attr <EOL> query = { query_key : identifier } <EOL> rv = self . user_model . objects ( ** query ) . first ( ) <EOL> if rv is not None : <EOL> return rv <EOL> def find_user ( self , ** kwargs ) : <EOL> try : <EOL> from mongoengine . queryset import Q , QCombination <EOL> except ImportError : <EOL> from mongoengine . queryset . visitor import Q , QCombination <EOL> from mongoengine . errors import ValidationError <EOL> queries = map ( lambda i : Q ( ** { i [ <NUM_LIT:0> ] : i [ <NUM_LIT:1> ] } ) , kwargs . items ( ) ) <EOL> query = QCombination ( QCombination . AND , queries ) <EOL> try : <EOL> return self . user_model . objects ( query ) . first ( ) <EOL> except ValidationError : <EOL> return None <EOL> def find_role ( self , role ) : <EOL> return self . role_model . objects ( name = role ) . first ( ) <EOL> class PeeweeUserDatastore ( PeeweeDatastore , UserDatastore ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , db , user_model , role_model , role_link ) : <EOL> PeeweeDatastore . __init__ ( self , db ) <EOL> UserDatastore . __init__ ( self , user_model , role_model ) <EOL> self . UserRole = role_link <EOL> def get_user ( self , identifier ) : <EOL> try : <EOL> return self . user_model . get ( self . user_model . id == identifier ) <EOL> except ValueError : <EOL> pass <EOL> for attr in get_identity_attributes ( ) : <EOL> column = getattr ( self . user_model , attr ) <EOL> try : <EOL> return self . user_model . get ( column ** identifier ) <EOL> except self . user_model . DoesNotExist : <EOL> pass <EOL> def find_user ( self , ** kwargs ) : <EOL> try : <EOL> return self . user_model . filter ( ** kwargs ) . get ( ) <EOL> except self . user_model . DoesNotExist : <EOL> return None <EOL> def find_role ( self , role ) : <EOL> try : <EOL> return self . role_model . filter ( name = role ) . get ( ) <EOL> except self . role_model . DoesNotExist : <EOL> return None <EOL> def create_user ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> roles = kwargs . pop ( '<STR_LIT>' , [ ] ) <EOL> user = self . user_model ( ** self . _prepare_create_user_args ( ** kwargs ) ) <EOL> user = self . put ( user ) <EOL> for role in roles : <EOL> self . add_role_to_user ( user , role ) <EOL> self . put ( user ) <EOL> return user <EOL> def add_role_to_user ( self , user , role ) : <EOL> """<STR_LIT>""" <EOL> user , role = self . _prepare_role_modify_args ( user , role ) <EOL> result = self . UserRole . select ( ) . where ( self . UserRole . user == user . id , self . UserRole . role == role . id ) <EOL> if result . count ( ) : <EOL> return False <EOL> else : <EOL> self . put ( self . UserRole . create ( user = user . id , role = role . id ) ) <EOL> return True <EOL> def remove_role_from_user ( self , user , role ) : <EOL> """<STR_LIT>""" <EOL> user , role = self . _prepare_role_modify_args ( user , role ) <EOL> result = self . UserRole . select ( ) . where ( self . UserRole . user == user , self . UserRole . role == role ) <EOL> if result . count ( ) : <EOL> query = self . UserRole . delete ( ) . where ( <EOL> self . UserRole . user == user , self . UserRole . role == role ) <EOL> query . execute ( ) <EOL> return True <EOL> else : <EOL> return False </s>
