<s> import six <EOL> from sqlalchemy . orm . collections import InstrumentedList <EOL> from werkzeug . exceptions import Forbidden <EOL> from werkzeug . utils import cached_property <EOL> from flask_principal import Permission , RoleNeed <EOL> from flask_potion . manager import RelationalManager <EOL> from flask_potion . fields import ToOne <EOL> from flask_potion . instances import Pagination <EOL> from . permission import HybridPermission <EOL> from . needs import HybridItemNeed , HybridUserNeed <EOL> PERMISSION_DEFAULTS = ( <EOL> ( '<STR_LIT>' , '<STR_LIT:yes>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ) <EOL> DEFAULT_METHODS = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> METHOD_ROUTE_RELATIONS = ( <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT>' ) ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , ) ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , ) ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , ) ) <EOL> ) <EOL> PERMISSION_DENIED_STRINGS = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> PERMISSION_GRANTED_STRINGS = ( '<STR_LIT:yes>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> class PrincipalMixin ( object ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( PrincipalMixin , self ) . __init__ ( * args , ** kwargs ) <EOL> raw_needs = dict ( PERMISSION_DEFAULTS ) <EOL> raw_needs . update ( self . resource . meta . get ( '<STR_LIT>' , { } ) ) <EOL> self . _raw_needs = raw_needs <EOL> @ cached_property <EOL> def _needs ( self ) : <EOL> needs_map = self . _raw_needs . copy ( ) <EOL> methods = needs_map . keys ( ) <EOL> def convert ( method , needs , map , path = ( ) ) : <EOL> options = set ( ) <EOL> if isinstance ( needs , six . string_types ) : <EOL> needs = [ needs ] <EOL> if isinstance ( needs , set ) : <EOL> return needs <EOL> for need in needs : <EOL> if need in PERMISSION_DENIED_STRINGS : <EOL> options . add ( Permission ( ( '<STR_LIT>' , ) ) ) <EOL> elif need in PERMISSION_GRANTED_STRINGS : <EOL> return { True } <EOL> elif need in methods : <EOL> if need == method : <EOL> options . add ( HybridItemNeed ( method , self . resource ) ) <EOL> elif need in path : <EOL> raise RuntimeError ( '<STR_LIT>' . format ( self . resource , path ) ) <EOL> else : <EOL> path += ( method , ) <EOL> options |= convert ( need , map [ need ] , map , path ) <EOL> elif '<STR_LIT::>' in need : <EOL> role , value = need . split ( '<STR_LIT::>' ) <EOL> field = self . resource . schema . fields [ value ] <EOL> if field . attribute is None : <EOL> field . attribute = value <EOL> if isinstance ( field , ToOne ) : <EOL> target = field . target <EOL> if role == '<STR_LIT:user>' : <EOL> options . add ( HybridUserNeed ( field ) ) <EOL> elif role == '<STR_LIT>' : <EOL> options . add ( RoleNeed ( value ) ) <EOL> else : <EOL> for imported_need in target . manager . _needs [ role ] : <EOL> if isinstance ( imported_need , HybridItemNeed ) : <EOL> imported_need = imported_need . extend ( field ) <EOL> options . add ( imported_need ) <EOL> else : <EOL> options . add ( RoleNeed ( need ) ) <EOL> return options <EOL> for method , needs in needs_map . items ( ) : <EOL> converted_needs = convert ( method , needs , needs_map ) <EOL> needs_map [ method ] = converted_needs <EOL> return needs_map <EOL> @ cached_property <EOL> def _permissions ( self ) : <EOL> permissions = { } <EOL> for method , needs in self . _needs . items ( ) : <EOL> if True in needs : <EOL> needs = set ( ) <EOL> permissions [ method ] = HybridPermission ( * needs ) <EOL> return permissions <EOL> def get_permissions_for_item ( self , item ) : <EOL> """<STR_LIT>""" <EOL> return { operation : permission . can ( item ) for operation , permission in self . _permissions . items ( ) } <EOL> def can_create_item ( self , item ) : <EOL> """<STR_LIT>""" <EOL> permission = self . _permissions [ '<STR_LIT>' ] <EOL> return permission . can ( item ) <EOL> def can_update_item ( self , item , changes = None ) : <EOL> """<STR_LIT>""" <EOL> permission = self . _permissions [ '<STR_LIT>' ] <EOL> return permission . can ( item ) <EOL> def can_delete_item ( self , item ) : <EOL> """<STR_LIT>""" <EOL> permission = self . _permissions [ '<STR_LIT>' ] <EOL> return permission . can ( item ) <EOL> def _query_filter_read_permission ( self , query ) : <EOL> read_permission = self . _permissions [ '<STR_LIT>' ] <EOL> return self . _query_filter_permission ( query , read_permission ) <EOL> def _query_filter_permission ( self , query , permission ) : <EOL> if permission . can ( ) : <EOL> return query <EOL> if not permission . hybrid_needs : <EOL> return None <EOL> expressions = [ ] <EOL> for need in permission . hybrid_needs : <EOL> ids = list ( need . identity_get_item_needs ( ) ) <EOL> if not ids : <EOL> continue <EOL> if len ( need . fields ) == <NUM_LIT:0> : <EOL> expression = self . _expression_for_ids ( ids ) <EOL> else : <EOL> expression = need . fields [ - <NUM_LIT:1> ] . target . manager . _expression_for_ids ( ids ) <EOL> for field in reversed ( need . fields ) : <EOL> expression = field . resource . manager . _expression_for_join ( field . attribute , expression ) <EOL> expressions . append ( expression ) <EOL> if not expressions : <EOL> return None <EOL> return self . _query_filter ( query , self . _or_expression ( expressions ) ) <EOL> def _query ( self , ** kwargs ) : <EOL> query = super ( PrincipalMixin , self ) . _query ( ** kwargs ) <EOL> read_permission = self . _permissions [ '<STR_LIT>' ] <EOL> query = self . _query_filter_permission ( query , read_permission ) <EOL> if query is None : <EOL> if all ( need . method == '<STR_LIT>' for need in read_permission . needs ) : <EOL> raise Forbidden ( ) <EOL> return query <EOL> def relation_instances ( self , item , attribute , target_resource , page = None , per_page = None ) : <EOL> query = getattr ( item , attribute ) <EOL> if isinstance ( query , InstrumentedList ) : <EOL> if page and per_page : <EOL> return Pagination . from_list ( query , page , per_page ) <EOL> return query <EOL> target_manager = target_resource . manager <EOL> if isinstance ( target_manager , PrincipalMixin ) : <EOL> query = target_manager . _query_filter_read_permission ( query ) <EOL> if page and per_page : <EOL> return target_manager . _query_get_paginated_items ( query , page , per_page ) <EOL> return target_manager . _query_get_all ( query ) <EOL> def create ( self , properties , commit = True ) : <EOL> if not self . can_create_item ( properties ) : <EOL> raise Forbidden ( ) <EOL> return super ( PrincipalMixin , self ) . create ( properties , commit ) <EOL> def update ( self , item , changes , * args , ** kwargs ) : <EOL> if not self . can_update_item ( item , changes ) : <EOL> raise Forbidden ( ) <EOL> return super ( PrincipalMixin , self ) . update ( item , changes , * args , ** kwargs ) <EOL> def delete ( self , item ) : <EOL> if not self . can_delete_item ( item ) : <EOL> raise Forbidden ( ) <EOL> return super ( PrincipalMixin , self ) . delete ( item ) <EOL> def principals ( manager ) : <EOL> if not issubclass ( manager , RelationalManager ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> class PrincipalsManager ( PrincipalMixin , manager ) : <EOL> pass <EOL> return PrincipalsManager </s>
