<s> from flask import current_app <EOL> from flask_sqlalchemy import Pagination as SAPagination , get_state <EOL> from sqlalchemy import String , or_ , and_ <EOL> from sqlalchemy . dialects import postgresql <EOL> from sqlalchemy . exc import IntegrityError <EOL> from sqlalchemy . orm import class_mapper , aliased <EOL> from sqlalchemy . orm . attributes import ScalarObjectAttributeImpl <EOL> from sqlalchemy . orm . collections import InstrumentedList <EOL> from sqlalchemy . orm . exc import NoResultFound <EOL> from flask_potion import fields <EOL> from flask_potion . contrib . alchemy . filters import FILTER_NAMES , FILTERS_BY_TYPE , SQLAlchemyBaseFilter <EOL> from flask_potion . exceptions import ItemNotFound , DuplicateKey , BackendConflict <EOL> from flask_potion . instances import Pagination <EOL> from flask_potion . manager import RelationalManager <EOL> from flask_potion . signals import before_add_to_relation , after_add_to_relation , before_remove_from_relation , after_remove_from_relation , before_create , after_create , before_update , after_update , before_delete , after_delete <EOL> from flask_potion . utils import get_value <EOL> class SQLAlchemyManager ( RelationalManager ) : <EOL> """<STR_LIT>""" <EOL> FILTER_NAMES = FILTER_NAMES <EOL> FILTERS_BY_TYPE = FILTERS_BY_TYPE <EOL> PAGINATION_TYPES = ( Pagination , SAPagination ) <EOL> def __init__ ( self , resource , model ) : <EOL> super ( SQLAlchemyManager , self ) . __init__ ( resource , model ) <EOL> def _init_model ( self , resource , model , meta ) : <EOL> mapper = class_mapper ( model ) <EOL> self . model = model <EOL> if meta . id_attribute : <EOL> self . id_column = getattr ( model , resource . meta . id_attribute ) <EOL> self . id_attribute = meta . id_attribute <EOL> else : <EOL> self . id_column = mapper . primary_key [ <NUM_LIT:0> ] <EOL> self . id_attribute = mapper . primary_key [ <NUM_LIT:0> ] . name <EOL> self . id_field = self . _get_field_from_column_type ( self . id_column , self . id_attribute , io = "<STR_LIT:r>" ) <EOL> fs = resource . schema <EOL> if meta . include_id : <EOL> fs . set ( '<STR_LIT>' , self . id_field ) <EOL> else : <EOL> fs . set ( '<STR_LIT>' , fields . ItemUri ( resource , attribute = self . id_attribute ) ) <EOL> if meta . include_type : <EOL> fs . set ( '<STR_LIT>' , fields . ItemType ( resource ) ) <EOL> if not hasattr ( resource . Meta , '<STR_LIT:name>' ) : <EOL> meta [ '<STR_LIT:name>' ] = model . __tablename__ . lower ( ) <EOL> fs = resource . schema <EOL> include_fields = meta . get ( '<STR_LIT>' , None ) <EOL> exclude_fields = meta . get ( '<STR_LIT>' , None ) <EOL> read_only_fields = meta . get ( '<STR_LIT>' , ( ) ) <EOL> write_only_fields = meta . get ( '<STR_LIT>' , ( ) ) <EOL> pre_declared_fields = { f . attribute or k for k , f in fs . fields . items ( ) } <EOL> for name , column in mapper . columns . items ( ) : <EOL> if ( include_fields and name in include_fields ) or ( exclude_fields and name not in exclude_fields ) or not ( include_fields or exclude_fields ) : <EOL> if column . primary_key or column . foreign_keys : <EOL> continue <EOL> if name in pre_declared_fields : <EOL> continue <EOL> io = "<STR_LIT>" <EOL> if name in read_only_fields : <EOL> io = "<STR_LIT:r>" <EOL> elif name in write_only_fields : <EOL> io = "<STR_LIT:w>" <EOL> if not ( column . nullable or column . default ) : <EOL> fs . required . add ( name ) <EOL> fs . set ( name , self . _get_field_from_column_type ( column , name , io = io ) ) <EOL> def _get_field_from_column_type ( self , column , attribute , io = "<STR_LIT>" ) : <EOL> args = ( ) <EOL> kwargs = { } <EOL> if isinstance ( column . type , postgresql . ARRAY ) : <EOL> field_class = fields . Array <EOL> args = ( fields . String , ) <EOL> elif isinstance ( column . type , postgresql . UUID ) : <EOL> field_class = fields . UUID <EOL> elif isinstance ( column . type , String ) and column . type . length : <EOL> field_class = fields . String <EOL> kwargs = { '<STR_LIT:max_length>' : column . type . length } <EOL> elif isinstance ( column . type , postgresql . HSTORE ) : <EOL> field_class = fields . Object <EOL> args = ( fields . String , ) <EOL> elif hasattr ( postgresql , '<STR_LIT>' ) and isinstance ( column . type , ( postgresql . JSON , postgresql . JSONB ) ) : <EOL> field_class = fields . Raw <EOL> kwargs = { "<STR_LIT>" : { } } <EOL> else : <EOL> field_class = self . _get_field_from_python_type ( column . type . python_type ) <EOL> kwargs [ '<STR_LIT>' ] = column . nullable <EOL> if column . default is not None and column . default . is_scalar : <EOL> kwargs [ '<STR_LIT:default>' ] = column . default . arg <EOL> return field_class ( * args , io = io , attribute = attribute , ** kwargs ) <EOL> def _init_filter ( self , filter_class , name , field , attribute ) : <EOL> return filter_class ( name , <EOL> field = field , <EOL> attribute = field . attribute or attribute , <EOL> column = getattr ( self . model , field . attribute or attribute ) ) <EOL> def _is_sortable_field ( self , field ) : <EOL> if super ( SQLAlchemyManager , self ) . _is_sortable_field ( field ) : <EOL> return True <EOL> elif isinstance ( field , fields . ToOne ) : <EOL> return isinstance ( field . target . manager , SQLAlchemyManager ) <EOL> else : <EOL> return False <EOL> @ staticmethod <EOL> def _get_session ( ) : <EOL> return get_state ( current_app ) . db . session <EOL> def _query ( self ) : <EOL> return self . model . query <EOL> def _query_filter ( self , query , expression ) : <EOL> return query . filter ( expression ) <EOL> def _expression_for_join ( self , attribute , expression ) : <EOL> relationship = getattr ( self . model , attribute ) <EOL> if isinstance ( relationship . impl , ScalarObjectAttributeImpl ) : <EOL> return relationship . has ( expression ) <EOL> else : <EOL> return relationship . any ( expression ) <EOL> def _expression_for_condition ( self , condition ) : <EOL> return condition . filter . expression ( condition . value ) <EOL> def _expression_for_ids ( self , ids ) : <EOL> return self . id_column . in_ ( ids ) <EOL> def _or_expression ( self , expressions ) : <EOL> if not expressions : <EOL> return True <EOL> if len ( expressions ) == <NUM_LIT:1> : <EOL> return expressions [ <NUM_LIT:0> ] <EOL> return or_ ( * expressions ) <EOL> def _and_expression ( self , expressions ) : <EOL> if not expressions : <EOL> return False <EOL> if len ( expressions ) == <NUM_LIT:1> : <EOL> return expressions [ <NUM_LIT:0> ] <EOL> return and_ ( * expressions ) <EOL> def _query_filter_by_id ( self , query , id ) : <EOL> try : <EOL> return query . filter ( self . id_column == id ) . one ( ) <EOL> except NoResultFound : <EOL> raise ItemNotFound ( self . resource , id = id ) <EOL> def _query_order_by ( self , query , sort ) : <EOL> order_clauses = [ ] <EOL> for field , attribute , reverse in sort : <EOL> column = getattr ( self . model , attribute ) <EOL> if isinstance ( field , fields . ToOne ) : <EOL> target_alias = aliased ( field . target . meta . model ) <EOL> query = query . outerjoin ( target_alias , column ) . reset_joinpoint ( ) <EOL> column = getattr ( target_alias , field . target . meta . sort_attribute or field . target . manager . id_attribute ) <EOL> order_clauses . append ( column . desc ( ) if reverse else column . asc ( ) ) <EOL> return query . order_by ( * order_clauses ) <EOL> def _query_get_paginated_items ( self , query , page , per_page ) : <EOL> return query . paginate ( page = page , per_page = per_page ) <EOL> def _query_get_all ( self , query ) : <EOL> return query . all ( ) <EOL> def _query_get_one ( self , query ) : <EOL> return query . one ( ) <EOL> def _query_get_first ( self , query ) : <EOL> try : <EOL> return query . one ( ) <EOL> except NoResultFound : <EOL> raise IndexError ( ) <EOL> def create ( self , properties , commit = True ) : <EOL> item = self . model ( ) <EOL> for key , value in properties . items ( ) : <EOL> setattr ( item , key , value ) <EOL> before_create . send ( self . resource , item = item ) <EOL> session = self . _get_session ( ) <EOL> try : <EOL> session . add ( item ) <EOL> if commit : <EOL> session . commit ( ) <EOL> except IntegrityError as e : <EOL> session . rollback ( ) <EOL> if hasattr ( e . orig , '<STR_LIT>' ) : <EOL> if e . orig . pgcode == "<STR_LIT>" : <EOL> raise DuplicateKey ( detail = e . orig . diag . message_detail ) <EOL> if current_app . debug : <EOL> raise BackendConflict ( debug_info = dict ( statement = e . statement , params = e . params ) ) <EOL> raise BackendConflict ( ) <EOL> after_create . send ( self . resource , item = item ) <EOL> return item <EOL> def update ( self , item , changes , commit = True ) : <EOL> session = self . _get_session ( ) <EOL> actual_changes = { <EOL> key : value for key , value in changes . items ( ) <EOL> if get_value ( key , item , None ) != value <EOL> } <EOL> try : <EOL> before_update . send ( self . resource , item = item , changes = actual_changes ) <EOL> for key , value in changes . items ( ) : <EOL> setattr ( item , key , value ) <EOL> if commit : <EOL> session . commit ( ) <EOL> except IntegrityError as e : <EOL> session . rollback ( ) <EOL> if hasattr ( e . orig , '<STR_LIT>' ) : <EOL> if e . orig . pgcode == '<STR_LIT>' : <EOL> raise DuplicateKey ( detail = e . orig . diag . message_detail ) <EOL> raise <EOL> after_update . send ( self . resource , item = item , changes = actual_changes ) <EOL> return item <EOL> def delete ( self , item ) : <EOL> before_delete . send ( self . resource , item = item ) <EOL> session = self . _get_session ( ) <EOL> session . delete ( item ) <EOL> session . commit ( ) <EOL> after_delete . send ( self . resource , item = item ) <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> if page and per_page : <EOL> return self . _query_get_paginated_items ( query , page , per_page ) <EOL> return self . _query_get_all ( query ) <EOL> def relation_add ( self , item , attribute , target_resource , target_item ) : <EOL> before_add_to_relation . send ( self . resource , item = item , attribute = attribute , child = target_item ) <EOL> getattr ( item , attribute ) . append ( target_item ) <EOL> after_add_to_relation . send ( self . resource , item = item , attribute = attribute , child = target_item ) <EOL> def relation_remove ( self , item , attribute , target_resource , target_item ) : <EOL> before_remove_from_relation . send ( self . resource , item = item , attribute = attribute , child = target_item ) <EOL> try : <EOL> getattr ( item , attribute ) . remove ( target_item ) <EOL> after_remove_from_relation . send ( self . resource , item = item , attribute = attribute , child = target_item ) <EOL> except ValueError : <EOL> pass <EOL> def commit ( self ) : <EOL> session = self . _get_session ( ) <EOL> session . commit ( ) </s>
