signature:
def cleanup ( self ) :
    """ensures that no data leaks from drop after processing by
 removing all data except the status file"""

implementation:
try :
    remove ( join ( self . fs_path , u'message' ) )
    remove ( join ( self . fs_path , 'dirty.zip.pgp' ) )
except OSError :
    pass
shutil . rmtree ( join ( self . fs_path , u'clean' ) , ignore_errors = True )
shutil . rmtree ( join ( self . fs_path , u'attach' ) , ignore_errors = True )


signature:
def _create_encrypted_zip ( self , source = 'dirty' , fs_target_dir = None ) :
    """creates a zip file from the drop and encrypts it to the editors .
 the encrypted archive is created inside fs _ target _ dir"""

implementation:
backup_recipients = [ r for r in self . editors if checkRecipient ( self . gpg_context , r ) ]
# this will be handled by watchdog , no need to send for each drop
if not backup_recipients :
    self . status = u'500 no valid keys at all'
    return self . status
# calculate paths
fs_backup = join ( self . fs_path , '%s.zip' % source )
if fs_target_dir is None :
    fs_backup_pgp = join ( self . fs_path , '%s.zip.pgp' % source )
else :
    fs_backup_pgp = join ( fs_target_dir , '%s.zip.pgp' % self . drop_id )
fs_source = dict ( dirty = self . fs_dirty_attachments , clean = self . fs_cleansed_attachments )
# create archive
with ZipFile ( fs_backup , 'w' , ZIP_STORED ) as backup :
    if exists ( join ( self . fs_path , 'message' ) ) :
        backup . write ( join ( self . fs_path , 'message' ) , arcname = 'message' )
    for fs_attachment in fs_source [ source ] :
        backup . write ( fs_attachment , arcname = split ( fs_attachment ) [ - 1 ] )
# encrypt archive
with open ( fs_backup , "rb" ) as backup :
    self . gpg_context . encrypt_file ( backup , backup_recipients , always_trust = True , output = fs_backup_pgp )
# cleanup
remove ( fs_backup )
return fs_backup_pgp


signature:
def _create_archive ( self ) :
    """creates an encrypted archive of the dropbox outside of the drop directory ."""

implementation:
self . status = u'270 creating final encrypted backup of cleansed attachments'
return self . _create_encrypted_zip ( source = 'clean' , fs_target_dir = self . container . fs_archive_cleansed )


signature:
def size_attachments ( self ) :
    """returns the number of bytes that the cleansed attachments take up on disk"""

implementation:
total_size = 0
for attachment in self . fs_cleansed_attachments :
    total_size += stat ( attachment ) . st_size
return total_size


signature:
def replies ( self ) :
    """returns a list of strings"""

implementation:
fs_reply_path = join ( self . fs_replies_path , 'message_001.txt' )
if exists ( fs_reply_path ) :
    return [ load ( open ( fs_reply_path , 'r' ) ) ]
else :
    return [ ]


signature:
def message ( self ) :
    """returns the user submitted text"""

implementation:
try :
    with open ( join ( self . fs_path , u'message' ) ) as message_file :
        return u'' . join ( [ line . decode ( 'utf-8' ) for line in message_file . readlines ( ) ] )
except IOError :
    return u''


signature:
def fs_dirty_attachments ( self ) :
    """returns a list of absolute paths to the attachements"""

implementation:
if exists ( self . fs_attachment_container ) :
    return [ join ( self . fs_attachment_container , attachment ) for attachment in listdir ( self . fs_attachment_container ) ]
else :
    return [ ]


signature:
def fs_cleansed_attachments ( self ) :
    """returns a list of absolute paths to the cleansed attachements"""

implementation:
if exists ( self . fs_cleansed_attachment_container ) :
    return [ join ( self . fs_cleansed_attachment_container , attachment ) for attachment in listdir ( self . fs_cleansed_attachment_container ) ]
else :
    return [ ]


signature:
def reset_cleansers ( confirm = True ) :
    """destroys all cleanser slaves and their rollback snapshots , as well as the initial master
 snapshot - this allows re - running the jailhost deployment to recreate fresh cleansers ."""

implementation:
if value_asbool ( confirm ) and not yesno ( """\nObacht!
            This will destroy any existing and or currently running cleanser jails.
            Are you sure that you want to continue?""" ) :
    exit ( "Glad I asked..." )
get_vars ( )
cleanser_count = AV [ 'ploy_cleanser_count' ]
# make sure no workers interfere :
fab . run ( 'ezjail-admin stop worker' )
# stop and nuke the cleanser slaves
for cleanser_index in range ( cleanser_count ) :
    cindex = '{:02d}' . format ( cleanser_index + 1 )
    fab . run ( 'ezjail-admin stop cleanser_{cindex}' . format ( cindex = cindex ) )
    with fab . warn_only ( ) :
        fab . run ( 'zfs destroy tank/jails/cleanser_{cindex}@jdispatch_rollback' . format ( cindex = cindex ) )
        fab . run ( 'ezjail-admin delete -fw cleanser_{cindex}' . format ( cindex = cindex ) )
        fab . run ( 'umount -f /usr/jails/cleanser_{cindex}' . format ( cindex = cindex ) )
        fab . run ( 'rm -rf /usr/jails/cleanser_{cindex}' . format ( cindex = cindex ) )
with fab . warn_only ( ) : # remove master snapshot
    fab . run ( 'zfs destroy -R tank/jails/cleanser@clonesource' )
    # restart worker and cleanser to prepare for subsequent ansible configuration runs
    fab . run ( 'ezjail-admin start worker' )
    fab . run ( 'ezjail-admin stop cleanser' )
    fab . run ( 'ezjail-admin start cleanser' )


signature:
def reset_jails ( confirm = True , keep_cleanser_master = True ) :
    """stops , deletes and re - creates all jails .
 since the cleanser master is rather large , that one is omitted by default ."""

implementation:
if value_asbool ( confirm ) and not yesno ( """\nObacht!
            This will destroy all existing and or currently running jails on the host.
            Are you sure that you want to continue?""" ) :
    exit ( "Glad I asked..." )
reset_cleansers ( confirm = False )
jails = [ 'appserver' , 'webserver' , 'worker' ]
if not value_asbool ( keep_cleanser_master ) :
    jails . append ( 'cleanser' )
with fab . warn_only ( ) :
    for jail in jails :
        fab . run ( 'ezjail-admin delete -fw {jail}' . format ( jail = jail ) )
    # remove authorized keys for no longer existing key ( they are regenerated for each new worker )
    fab . run ( 'rm /usr/jails/cleanser/usr/home/cleanser/.ssh/authorized_keys' )


signature:
def dict_dumper ( provider ) :
    '''Dump a provider to a format that can be passed to a
 : class : ` skosprovider . providers . DictionaryProvider ` .
 : param skosprovider . providers . VocabularyProvider provider : The provider
 that wil be turned into a ` dict ` .
 : rtype : A list of dicts .
 . . versionadded : : 0.2.0'''

implementation:
ret = [ ]
for stuff in provider . get_all ( ) :
    c = provider . get_by_id ( stuff [ 'id' ] )
    labels = [ l . __dict__ for l in c . labels ]
    notes = [ n . __dict__ for n in c . notes ]
    sources = [ s . __dict__ for s in c . sources ]
    if isinstance ( c , Concept ) :
        ret . append ( { 'id' : c . id , 'uri' : c . uri , 'type' : c . type , 'labels' : labels , 'notes' : notes , 'sources' : sources , 'narrower' : c . narrower , 'broader' : c . broader , 'related' : c . related , 'member_of' : c . member_of , 'subordinate_arrays' : c . subordinate_arrays , 'matches' : c . matches } )
    elif isinstance ( c , Collection ) :
        ret . append ( { 'id' : c . id , 'uri' : c . uri , 'type' : c . type , 'labels' : labels , 'notes' : notes , 'sources' : sources , 'members' : c . members , 'member_of' : c . member_of , 'superordinates' : c . superordinates } )
return ret


signature:
def _add_admin ( self , app , ** kwargs ) :
    """Add a Flask Admin interface to an application .
 : param flask . Flask app : A Flask application
 : param kwargs : Keyword arguments are passed through to : class : ` flask _ admin . Admin `
 : rtype : flask _ admin . Admin"""

implementation:
from flask_admin import Admin
from flask_admin . contrib . sqla import ModelView
admin = Admin ( app , ** kwargs )
for flask_admin_model in self . flask_admin_models :
    if isinstance ( flask_admin_model , tuple ) : # assume its a 2 tuple
        if len ( flask_admin_model ) != 2 :
            raise TypeError
        model , view = flask_admin_model
        admin . add_view ( view ( model , self . session ) )
    else :
        admin . add_view ( ModelView ( flask_admin_model , self . session ) )
return admin


signature:
def get_flask_admin_app ( self , url : Optional [ str ] = None , secret_key : Optional [ str ] = None ) :
    """Create a Flask application .
 : param url : Optional mount point of the admin application . Defaults to ` ` ' / ' ` ` .
 : rtype : flask . Flask"""

implementation:
from flask import Flask
app = Flask ( __name__ )
if secret_key :
    app . secret_key = secret_key
self . _add_admin ( app , url = ( url or '/' ) )
return app


signature:
def get_cli ( cls ) -> click . Group :
    """Add a : mod : ` click ` main function to use as a command line interface ."""

implementation:
main = super ( ) . get_cli ( )
cls . _cli_add_flask ( main )
return main


signature:
def _handle_system_status_event ( self , event : SystemStatusEvent ) -> None :
    """DISARMED - > ARMED _ AWAY - > EXIT _ DELAY _ START - > EXIT _ DELAY _ END
 ( trip ) : - > ALARM - > OUTPUT _ ON - > ALARM _ RESTORE
 ( disarm ) : - > DISARMED - > OUTPUT _ OFF
 ( disarm ) : - > DISARMED
 ( disarm before EXIT _ DELAY _ END ) : - > DISARMED - > EXIT _ DELAY _ END
 TODO ( NW ) : Check ALARM _ RESTORE state transition to move back into ARMED _ AWAY state"""

implementation:
if event . type == SystemStatusEvent . EventType . UNSEALED :
    return self . _update_zone ( event . zone , True )
elif event . type == SystemStatusEvent . EventType . SEALED :
    return self . _update_zone ( event . zone , False )
elif event . type == SystemStatusEvent . EventType . ALARM :
    return self . _update_arming_state ( ArmingState . TRIGGERED )
elif event . type == SystemStatusEvent . EventType . ALARM_RESTORE :
    if self . arming_state != ArmingState . DISARMED :
        return self . _update_arming_state ( ArmingState . ARMED )
elif event . type == SystemStatusEvent . EventType . ENTRY_DELAY_START :
    return self . _update_arming_state ( ArmingState . ENTRY_DELAY )
elif event . type == SystemStatusEvent . EventType . ENTRY_DELAY_END :
    pass
elif event . type == SystemStatusEvent . EventType . EXIT_DELAY_START :
    return self . _update_arming_state ( ArmingState . EXIT_DELAY )
elif event . type == SystemStatusEvent . EventType . EXIT_DELAY_END : # Exit delay finished - if we were in the process of arming update
# state to armed
    if self . arming_state == ArmingState . EXIT_DELAY :
        return self . _update_arming_state ( ArmingState . ARMED )
elif event . type in Alarm . ARM_EVENTS :
    return self . _update_arming_state ( ArmingState . ARMING )
elif event . type == SystemStatusEvent . EventType . DISARMED :
    return self . _update_arming_state ( ArmingState . DISARMED )
elif event . type == SystemStatusEvent . EventType . ARMING_DELAYED :
    pass


signature:
def dropbox_form ( request ) :
    """generates a dropbox uid and renders the submission form with a signed version of that id"""

implementation:
from briefkasten import generate_post_token
token = generate_post_token ( secret = request . registry . settings [ 'post_secret' ] )
return dict ( action = request . route_url ( 'dropbox_form_submit' , token = token ) , fileupload_url = request . route_url ( 'dropbox_fileupload' , token = token ) , ** defaults ( request ) )


signature:
def dropbox_fileupload ( dropbox , request ) :
    """accepts a single file upload and adds it to the dropbox as attachment"""

implementation:
attachment = request . POST [ 'attachment' ]
attached = dropbox . add_attachment ( attachment )
return dict ( files = [ dict ( name = attached , type = attachment . type , ) ] )


signature:
def dropbox_submission ( dropbox , request ) :
    """handles the form submission , redirects to the dropbox ' s status page ."""

implementation:
try :
    data = dropbox_schema . deserialize ( request . POST )
except Exception :
    return HTTPFound ( location = request . route_url ( 'dropbox_form' ) )
# set the message
dropbox . message = data . get ( 'message' )
# recognize submission from watchdog
if 'testing_secret' in dropbox . settings :
    dropbox . from_watchdog = is_equal ( unicode ( dropbox . settings [ 'test_submission_secret' ] ) , data . pop ( 'testing_secret' , u'' ) )
# a non - js client might have uploaded an attachment via the form ' s fileupload field :
if data . get ( 'upload' ) is not None :
    dropbox . add_attachment ( data [ 'upload' ] )
# now we can call the process method
dropbox . submit ( )
drop_url = request . route_url ( 'dropbox_view' , drop_id = dropbox . drop_id )
print ( "Created dropbox %s" % drop_url )
return HTTPFound ( location = drop_url )


signature:
def belns ( keyword : str , file : TextIO , encoding : Optional [ str ] , use_names : bool ) :
    """Write as a BEL namespace ."""

implementation:
directory = get_data_dir ( keyword )
obo_url = f'http://purl.obolibrary.org/obo/{keyword}.obo'
obo_path = os . path . join ( directory , f'{keyword}.obo' )
obo_cache_path = os . path . join ( directory , f'{keyword}.obo.pickle' )
obo_getter = make_obo_getter ( obo_url , obo_path , preparsed_path = obo_cache_path )
graph = obo_getter ( )
convert_obo_graph_to_belns ( graph , file = file , encoding = encoding , use_names = use_names , )


signature:
def belanno ( keyword : str , file : TextIO ) :
    """Write as a BEL annotation ."""

implementation:
directory = get_data_dir ( keyword )
obo_url = f'http://purl.obolibrary.org/obo/{keyword}.obo'
obo_path = os . path . join ( directory , f'{keyword}.obo' )
obo_cache_path = os . path . join ( directory , f'{keyword}.obo.pickle' )
obo_getter = make_obo_getter ( obo_url , obo_path , preparsed_path = obo_cache_path )
graph = obo_getter ( )
convert_obo_graph_to_belanno ( graph , file = file , )


signature:
def _store_helper ( model : Action , session : Optional [ Session ] = None ) -> None :
    """Help store an action ."""

implementation:
if session is None :
    session = _make_session ( )
session . add ( model )
session . commit ( )
session . close ( )


signature:
def _make_session ( connection : Optional [ str ] = None ) -> Session :
    """Make a session ."""

implementation:
if connection is None :
    connection = get_global_connection ( )
engine = create_engine ( connection )
create_all ( engine )
session_cls = sessionmaker ( bind = engine )
session = session_cls ( )
return session


signature:
def create_all ( engine , checkfirst = True ) :
    """Create the tables for Bio2BEL ."""

implementation: Base . metadata . create_all ( bind = engine , checkfirst = checkfirst )


signature:
def store_populate ( cls , resource : str , session : Optional [ Session ] = None ) -> 'Action' :
    """Store a " populate " event .
 : param resource : The normalized name of the resource to store
 Example :
 > > > from bio2bel . models import Action
 > > > Action . store _ populate ( ' hgnc ' )"""

implementation:
action = cls . make_populate ( resource )
_store_helper ( action , session = session )
return action


signature:
def store_populate_failed ( cls , resource : str , session : Optional [ Session ] = None ) -> 'Action' :
    """Store a " populate failed " event .
 : param resource : The normalized name of the resource to store
 Example :
 > > > from bio2bel . models import Action
 > > > Action . store _ populate _ failed ( ' hgnc ' )"""

implementation:
action = cls . make_populate_failed ( resource )
_store_helper ( action , session = session )
return action


signature:
def store_drop ( cls , resource : str , session : Optional [ Session ] = None ) -> 'Action' :
    """Store a " drop " event .
 : param resource : The normalized name of the resource to store
 Example :
 > > > from bio2bel . models import Action
 > > > Action . store _ drop ( ' hgnc ' )"""

implementation:
action = cls . make_drop ( resource )
_store_helper ( action , session = session )
return action


signature:
def ls ( cls , session : Optional [ Session ] = None ) -> List [ 'Action' ] :
    """Get all actions ."""

implementation:
if session is None :
    session = _make_session ( )
actions = session . query ( cls ) . order_by ( cls . created . desc ( ) ) . all ( )
session . close ( )
return actions


signature:
def count ( cls , session : Optional [ Session ] = None ) -> int :
    """Count all actions ."""

implementation:
if session is None :
    session = _make_session ( )
count = session . query ( cls ) . count ( )
session . close ( )
return count


signature:
def get_data_dir ( module_name : str ) -> str :
    """Ensure the appropriate Bio2BEL data directory exists for the given module , then returns the file path .
 : param module _ name : The name of the module . Ex : ' chembl '
 : return : The module ' s data directory"""

implementation:
module_name = module_name . lower ( )
data_dir = os . path . join ( BIO2BEL_DIR , module_name )
os . makedirs ( data_dir , exist_ok = True )
return data_dir


signature:
def get_module_config_cls ( module_name : str ) -> Type [ _AbstractModuleConfig ] : # noqa : D202
    """Build a module configuration class ."""

implementation:
class ModuleConfig ( _AbstractModuleConfig ) :
    NAME = f'bio2bel:{module_name}'
    FILES = DEFAULT_CONFIG_PATHS + [ os . path . join ( DEFAULT_CONFIG_DIRECTORY , module_name , 'config.ini' ) ]
return ModuleConfig


signature:
def get_connection ( module_name : str , connection : Optional [ str ] = None ) -> str :
    """Return the SQLAlchemy connection string if it is set .
 Order of operations :
 1 . Return the connection if given as a parameter
 2 . Check the environment for BIO2BEL _ { module _ name } _ CONNECTION
 3 . Look in the bio2bel config file for module - specific connection . Create if doesn ' t exist . Check the
 module - specific section for ` ` connection ` `
 4 . Look in the bio2bel module folder for a config file . Don ' t create if doesn ' t exist . Check the default section
 for ` ` connection ` `
 5 . Check the environment for BIO2BEL _ CONNECTION
 6 . Check the bio2bel config file for default
 7 . Fall back to standard default cache connection
 : param module _ name : The name of the module to get the configuration for
 : param connection : get the SQLAlchemy connection string
 : return : The SQLAlchemy connection string based on the configuration"""

implementation:
# 1 . Use given connection
if connection is not None :
    return connection
module_name = module_name . lower ( )
module_config_cls = get_module_config_cls ( module_name )
module_config = module_config_cls . load ( )
return module_config . connection or config . connection


signature:
def get_modules ( ) -> Mapping :
    """Get all Bio2BEL modules ."""

implementation:
modules = { }
for entry_point in iter_entry_points ( group = 'bio2bel' , name = None ) :
    entry = entry_point . name
    try :
        modules [ entry ] = entry_point . load ( )
    except VersionConflict as exc :
        log . warning ( 'Version conflict in %s: %s' , entry , exc )
        continue
    except UnknownExtra as exc :
        log . warning ( 'Unknown extra in %s: %s' , entry , exc )
        continue
    except ImportError as exc :
        log . exception ( 'Issue with importing module %s: %s' , entry , exc )
        continue
return modules


signature:
def clear_cache ( module_name : str , keep_database : bool = True ) -> None :
    """Clear all downloaded files ."""

implementation:
data_dir = get_data_dir ( module_name )
if not os . path . exists ( data_dir ) :
    return
for name in os . listdir ( data_dir ) :
    if name in { 'config.ini' , 'cfg.ini' } :
        continue
    if name == 'cache.db' and keep_database :
        continue
    path = os . path . join ( data_dir , name )
    if os . path . isdir ( path ) :
        shutil . rmtree ( path )
    else :
        os . remove ( path )
    os . rmdir ( data_dir )


signature:
def create_all ( self , check_first : bool = True ) :
    """Create the empty database ( tables ) .
 : param bool check _ first : Defaults to True , don ' t issue CREATEs for tables already present
 in the target database . Defers to : meth : ` sqlalchemy . sql . schema . MetaData . create _ all `"""

implementation: self . _metadata . create_all ( self . engine , checkfirst = check_first )


signature:
def drop_all ( self , check_first : bool = True ) :
    """Drop all tables from the database .
 : param bool check _ first : Defaults to True , only issue DROPs for tables confirmed to be
 present in the target database . Defers to : meth : ` sqlalchemy . sql . schema . MetaData . drop _ all `"""

implementation:
self . _metadata . drop_all ( self . engine , checkfirst = check_first )
self . _store_drop ( )


signature:
def get_cli ( cls ) -> click . Group :
    """Get the : mod : ` click ` main function to use as a command line interface ."""

implementation:
main = super ( ) . get_cli ( )
cls . _cli_add_populate ( main )
cls . _cli_add_drop ( main )
cls . _cli_add_cache ( main )
cls . _cli_add_summarize ( main )
return main


signature:
def label ( labels = [ ] , language = 'any' , sortLabel = False ) :
    '''Provide a label for a list of labels .
 The items in the list of labels are assumed to be either instances of
 : class : ` Label ` , or dicts with at least the key ` label ` in them . These will
 be passed to the : func : ` dict _ to _ label ` function .
 This method tries to find a label by looking if there ' s
 a pref label for the specified language . If there ' s no pref label ,
 it looks for an alt label . It disregards hidden labels .
 While matching languages , preference will be given to exact matches . But ,
 if no exact match is present , an inexact match will be attempted . This might
 be because a label in language ` nl - BE ` is being requested , but only ` nl ` or
 even ` nl - NL ` is present . Similarly , when requesting ` nl ` , a label with
 language ` nl - NL ` or even ` nl - Latn - NL ` will also be considered ,
 providing no label is present that has an exact match with the
 requested language .
 If language ' any ' was specified , all labels will be considered ,
 regardless of language .
 To find a label without a specified language , pass ` None ` as language .
 If a language or None was specified , and no label could be found , this
 method will automatically try to find a label in some other language .
 Finally , if no label could be found , None is returned .
 : param string language : The preferred language to receive the label in . This
 should be a valid IANA language tag .
 : param boolean sortLabel : Should sortLabels be considered or not ? If True ,
 sortLabels will be preferred over prefLabels . Bear in mind that these
 are still language dependent . So , it ' s possible to have a different
 sortLabel per language .
 : rtype : A : class : ` Label ` or ` None ` if no label could be found .'''

implementation:
if not labels :
    return None
if not language :
    language = 'und'
labels = [ dict_to_label ( l ) for l in labels ]
l = False
if sortLabel :
    l = find_best_label_for_type ( labels , language , 'sortLabel' )
if not l :
    l = find_best_label_for_type ( labels , language , 'prefLabel' )
if not l :
    l = find_best_label_for_type ( labels , language , 'altLabel' )
if l :
    return l
else :
    return label ( labels , 'any' , sortLabel ) if language != 'any' else None


signature:
def find_best_label_for_type ( labels , language , labeltype ) :
    '''Find the best label for a certain labeltype .
 : param list labels : A list of : class : ` Label ` .
 : param str language : An IANA language string , eg . ` nl ` or ` nl - BE ` .
 : param str labeltype : Type of label to look for , eg . ` prefLabel ` .'''

implementation:
typelabels = [ l for l in labels if l . type == labeltype ]
if not typelabels :
    return False
if language == 'any' :
    return typelabels [ 0 ]
exact = filter_labels_by_language ( typelabels , language )
if exact :
    return exact [ 0 ]
inexact = filter_labels_by_language ( typelabels , language , True )
if inexact :
    return inexact [ 0 ]
return False


signature:
def filter_labels_by_language ( labels , language , broader = False ) :
    '''Filter a list of labels , leaving only labels of a certain language .
 : param list labels : A list of : class : ` Label ` .
 : param str language : An IANA language string , eg . ` nl ` or ` nl - BE ` .
 : param boolean broader : When true , will also match ` nl - BE ` when filtering
 on ` nl ` . When false , only exact matches are considered .'''

implementation:
if language == 'any' :
    return labels
if broader :
    language = tags . tag ( language ) . language . format
    return [ l for l in labels if tags . tag ( l . language ) . language . format == language ]
else :
    language = tags . tag ( language ) . format
    return [ l for l in labels if tags . tag ( l . language ) . format == language ]


signature:
def dict_to_label ( dict ) :
    '''Transform a dict with keys ` label ` , ` type ` and ` language ` into a
 : class : ` Label ` .
 Only the ` label ` key is mandatory . If ` type ` is not present , it will
 default to ` prefLabel ` . If ` language ` is not present , it will default
 to ` und ` .
 If the argument passed is not a dict , this method just
 returns the argument .'''

implementation:
try :
    return Label ( dict [ 'label' ] , dict . get ( 'type' , 'prefLabel' ) , dict . get ( 'language' , 'und' ) )
except ( KeyError , AttributeError , TypeError ) :
    return dict


signature:
def dict_to_note ( dict ) :
    '''Transform a dict with keys ` note ` , ` type ` and ` language ` into a
 : class : ` Note ` .
 Only the ` note ` key is mandatory . If ` type ` is not present , it will
 default to ` note ` . If ` language ` is not present , it will default to ` und ` .
 If ` markup ` is not present it will default to ` None ` .
 If the argument passed is already a : class : ` Note ` , this method just returns
 the argument .'''

implementation:
if isinstance ( dict , Note ) :
    return dict
return Note ( dict [ 'note' ] , dict . get ( 'type' , 'note' ) , dict . get ( 'language' , 'und' ) , dict . get ( 'markup' ) )


signature:
def dict_to_source ( dict ) :
    '''Transform a dict with key ' citation ' into a : class : ` Source ` .
 If the argument passed is already a : class : ` Source ` , this method just
 returns the argument .'''

implementation:
if isinstance ( dict , Source ) :
    return dict
return Source ( dict [ 'citation' ] , dict . get ( 'markup' ) )


signature:
def _sortkey ( self , key = 'uri' , language = 'any' ) :
    '''Provide a single sortkey for this conceptscheme .
 : param string key : Either ` uri ` , ` label ` or ` sortlabel ` .
 : param string language : The preferred language to receive the label in
 if key is ` label ` or ` sortlabel ` . This should be a valid IANA language tag .
 : rtype : : class : ` str `'''

implementation:
if key == 'uri' :
    return self . uri
else :
    l = label ( self . labels , language , key == 'sortlabel' )
    return l . label . lower ( ) if l else ''


signature:
def _iterate_managers ( connection , skip ) :
    """Iterate over instantiated managers ."""

implementation:
for idx , name , manager_cls in _iterate_manage_classes ( skip ) :
    if name in skip :
        continue
    try :
        manager = manager_cls ( connection = connection )
    except TypeError as e :
        click . secho ( f'Could not instantiate {name}: {e}' , fg = 'red' )
    else :
        yield idx , name , manager


signature:
def populate ( connection , reset , force , skip ) :
    """Populate all ."""

implementation:
for idx , name , manager in _iterate_managers ( connection , skip ) :
    click . echo ( click . style ( f'[{idx}/{len(MANAGERS)}] ' , fg = 'blue' , bold = True ) + click . style ( f'populating {name}' , fg = 'cyan' , bold = True ) )
    if reset :
        click . echo ( 'deleting the previous instance of the database' )
        manager . drop_all ( )
        click . echo ( 'creating new models' )
        manager . create_all ( )
    elif manager . is_populated ( ) and not force :
        click . echo ( f'👍 {name} is already populated. use --force to overwrite' , color = 'red' )
        continue
    try :
        manager . populate ( )
    except Exception :
        logger . exception ( '%s population failed' , name )
        click . secho ( f'👎 {name} population failed' , fg = 'red' , bold = True )


signature:
def drop ( connection , skip ) :
    """Drop all ."""

implementation:
for idx , name , manager in _iterate_managers ( connection , skip ) :
    click . secho ( f'dropping {name}' , fg = 'cyan' , bold = True )
    manager . drop_all ( )


signature:
def clear ( skip ) :
    """Clear all caches ."""

implementation:
for name in sorted ( MODULES ) :
    if name in skip :
        continue
    click . secho ( f'clearing cache for {name}' , fg = 'cyan' , bold = True )
    clear_cache ( name )


signature:
def summarize ( connection , skip ) :
    """Summarize all ."""

implementation:
for idx , name , manager in _iterate_managers ( connection , skip ) :
    click . secho ( name , fg = 'cyan' , bold = True )
    if not manager . is_populated ( ) :
        click . echo ( '👎 unpopulated' )
        continue
    if isinstance ( manager , BELNamespaceManagerMixin ) :
        click . secho ( f'Terms: {manager._count_model(manager.namespace_model)}' , fg = 'green' )
    if isinstance ( manager , BELManagerMixin ) :
        try :
            click . secho ( f'Relations: {manager.count_relations()}' , fg = 'green' )
        except TypeError as e :
            click . secho ( str ( e ) , fg = 'red' )
    for field_name , count in sorted ( manager . summarize ( ) . items ( ) ) :
        click . echo ( click . style ( '=> ' , fg = 'white' , bold = True ) + f"{field_name.replace('_', ' ').capitalize()}: {count}" )


signature:
def sheet ( connection , skip , file : TextIO ) :
    """Generate a summary sheet ."""

implementation:
from tabulate import tabulate
header = [ '' , 'Name' , 'Description' , 'Terms' , 'Relations' ]
rows = [ ]
for i , ( idx , name , manager ) in enumerate ( _iterate_managers ( connection , skip ) , start = 1 ) :
    try :
        if not manager . is_populated ( ) :
            continue
    except AttributeError :
        click . secho ( f'{name} does not implement is_populated' , fg = 'red' )
        continue
    terms , relations = None , None
    if isinstance ( manager , BELNamespaceManagerMixin ) :
        terms = manager . _count_model ( manager . namespace_model )
    if isinstance ( manager , BELManagerMixin ) :
        try :
            relations = manager . count_relations ( )
        except TypeError as e :
            relations = str ( e )
    rows . append ( ( i , name , manager . __doc__ . split ( '\n' ) [ 0 ] . strip ( ) . strip ( '.' ) , terms , relations ) )
print ( tabulate ( rows , headers = header , # tablefmt = " fancy _ grid " ,
) )


signature:
def write ( connection , skip , directory , force ) :
    """Write a BEL namespace names / identifiers to terminology store ."""

implementation:
os . makedirs ( directory , exist_ok = True )
from . manager . namespace_manager import BELNamespaceManagerMixin
for idx , name , manager in _iterate_managers ( connection , skip ) :
    if not ( isinstance ( manager , AbstractManager ) and isinstance ( manager , BELNamespaceManagerMixin ) ) :
        continue
    click . secho ( name , fg = 'cyan' , bold = True )
    if force :
        try :
            click . echo ( f'dropping' )
            manager . drop_all ( )
            click . echo ( 'clearing cache' )
            clear_cache ( name )
            click . echo ( 'populating' )
            manager . populate ( )
        except Exception :
            click . secho ( f'{name} failed' , fg = 'red' )
            continue
    try :
        r = manager . write_directory ( directory )
    except TypeError as e :
        click . secho ( f'error with {name}: {e}' . rstrip ( ) , fg = 'red' )
    else :
        if not r :
            click . echo ( 'no update' )


signature:
def write ( connection , skip , directory , force ) :
    """Write all as BEL ."""

implementation:
os . makedirs ( directory , exist_ok = True )
from . manager . bel_manager import BELManagerMixin
import pybel
for idx , name , manager in _iterate_managers ( connection , skip ) :
    if not isinstance ( manager , BELManagerMixin ) :
        continue
    click . secho ( name , fg = 'cyan' , bold = True )
    path = os . path . join ( directory , f'{name}.bel.pickle' )
    if os . path . exists ( path ) and not force :
        click . echo ( '👍 already exported' )
        continue
    if not manager . is_populated ( ) :
        click . echo ( '👎 unpopulated' )
    else :
        graph = manager . to_bel ( )
        pybel . to_pickle ( graph , path )
        pybel . to_json_path ( graph , os . path . join ( directory , f'{name}.bel.json' ) )


signature:
def web ( connection , host , port ) :
    """Run a combine web interface ."""

implementation:
from bio2bel . web . application import create_application
app = create_application ( connection = connection )
app . run ( host = host , port = port )


signature:
def actions ( connection ) :
    """List all actions ."""

implementation:
session = _make_session ( connection = connection )
for action in Action . ls ( session = session ) :
    click . echo ( f'{action.created} {action.action} {action.resource}' )


signature:
def count_relations ( self ) -> int :
    """Count the number of BEL relations generated ."""

implementation:
if self . edge_model is ... :
    raise Bio2BELMissingEdgeModelError ( 'edge_edge model is undefined/count_bel_relations is not overridden' )
elif isinstance ( self . edge_model , list ) :
    return sum ( self . _count_model ( m ) for m in self . edge_model )
else :
    return self . _count_model ( self . edge_model )


signature:
def to_indra_statements ( self , * args , ** kwargs ) :
    """Dump as a list of INDRA statements .
 : rtype : List [ indra . Statement ]"""

implementation:
graph = self . to_bel ( * args , ** kwargs )
return to_indra_statements ( graph )


signature:
def exebench ( width ) :
    """benchorg . jpg is
 ' http : / / upload . wikimedia . org / wikipedia / commons / d / df / SAND _ LUE . jpg '"""

implementation:
height = width * 2 / 3
with Benchmarker ( width = 30 , loop = N ) as bm :
    for i in bm ( 'kaa.imlib2' ) :
        imlib2_scale ( 'benchorg.jpg' , width , height )
    for i in bm ( "PIL" ) :
        pil_scale ( 'benchorg.jpg' , width , height )
    for i in bm ( "pgmagick(blob-read)" ) :
        pgmagick_scale_from_blob ( 'benchorg.jpg' , width , height )
    for i in bm ( "pgmagick(normal-read)" ) :
        pgmagick_scale ( 'benchorg.jpg' , width , height )
    for i in bm ( "pgmagick(scale+sharpen)" ) :
        pgmagick_scale_plus_sharpen ( 'benchorg.jpg' , width , height )
    for i in bm ( "opencv" ) :
        opencv_scale ( 'benchorg.jpg' , width , height )
    for i in bm ( "pyimlib2" ) :
        pyimlib2_scale ( 'benchorg.jpg' , width , height )
    for i in bm ( "pyimlib2_with_pgsharpen" ) :
        pyimlib2_scale_with_pgmagicksharpen ( 'benchorg.jpg' , width , height )
return bm . results


signature:
def _convert_coordinatelist ( input_obj ) :
    """convert from ' list ' or ' tuple ' object to pgmagick . CoordinateList .
 : type input _ obj : list or tuple"""

implementation:
cdl = pgmagick . CoordinateList ( )
for obj in input_obj :
    cdl . append ( pgmagick . Coordinate ( obj [ 0 ] , obj [ 1 ] ) )
return cdl


signature:
def _convert_vpathlist ( input_obj ) :
    """convert from ' list ' or ' tuple ' object to pgmagick . VPathList .
 : type input _ obj : list or tuple"""

implementation:
vpl = pgmagick . VPathList ( )
for obj in input_obj : # FIXME
    obj = pgmagick . PathMovetoAbs ( pgmagick . Coordinate ( obj [ 0 ] , obj [ 1 ] ) )
    vpl . append ( obj )
return vpl


signature:
def get_exif_info ( self ) :
    """return exif - tag dict"""

implementation:
_dict = { }
for tag in _EXIF_TAGS :
    ret = self . img . attribute ( "EXIF:%s" % tag )
    if ret and ret != 'unknown' :
        _dict [ tag ] = ret
return _dict


signature:
def bezier ( self , points ) :
    """Draw a Bezier - curve .
 : param points : ex . ) ( ( 5 , 5 ) , ( 6 , 6 ) , ( 7 , 7 ) )
 : type points : list"""

implementation:
coordinates = pgmagick . CoordinateList ( )
for point in points :
    x , y = float ( point [ 0 ] ) , float ( point [ 1 ] )
    coordinates . append ( pgmagick . Coordinate ( x , y ) )
self . drawer . append ( pgmagick . DrawableBezier ( coordinates ) )


signature:
def color ( self , x , y , paint_method ) :
    """: param paint _ method : ' point ' or ' replace ' or ' floodfill ' or
 ' filltoborder ' or ' reset '
 : type paint _ method : str or pgmagick . PaintMethod"""

implementation:
paint_method = _convert_paintmethod ( paint_method )
color = pgmagick . DrawableColor ( x , y , paint_method )
self . drawer . append ( color )


signature:
def ellipse ( self , org_x , org_y , radius_x , radius_y , arc_start , arc_end ) :
    """: param org _ x : origination x axis
 : param org _ y : origination y axis
 : param radius _ x : radius x axis
 : param radius _ y : radius y axis
 : param arc _ start : arc start angle
 : param arc _ end : arc end angle"""

implementation:
ellipse = pgmagick . DrawableEllipse ( float ( org_x ) , float ( org_y ) , float ( radius_x ) , float ( radius_y ) , float ( arc_start ) , float ( arc_end ) )
self . drawer . append ( ellipse )


signature:
def fill_opacity ( self , opacity ) :
    """: param opacity : 0.0 ~ 1.0"""

implementation:
opacity = pgmagick . DrawableFillOpacity ( float ( opacity ) )
self . drawer . append ( opacity )


signature:
def matte ( self , x , y , paint_method ) :
    """: param paint _ method : ' point ' or ' replace ' or ' floodfill ' or
 ' filltoborder ' or ' reset '
 : type paint _ method : str or pgmagick . PaintMethod"""

implementation:
paint_method = _convert_paintmethod ( paint_method )
self . drawer . append ( pgmagick . DrawableMatte ( x , y , paint_method ) )


signature:
def scaling ( self , x , y ) :
    """Scaling Draw Object
 : param x : 0.0 ~ 1.0
 : param y : 0.0 ~ 1.0"""

implementation: self . drawer . append ( pgmagick . DrawableScaling ( float ( x ) , float ( y ) ) )


signature:
def stroke_antialias ( self , flag = True ) :
    """stroke antialias
 : param flag : True or False . ( default is True )
 : type flag : bool"""

implementation:
antialias = pgmagick . DrawableStrokeAntialias ( flag )
self . drawer . append ( antialias )


signature:
def stroke_linecap ( self , linecap ) :
    """set to stroke linecap .
 : param linecap : ' undefined ' , ' butt ' , ' round ' , ' square '
 : type linecap : str"""

implementation:
linecap = getattr ( pgmagick . LineCap , "%sCap" % linecap . title ( ) )
linecap = pgmagick . DrawableStrokeLineCap ( linecap )
self . drawer . append ( linecap )


signature:
def stroke_linejoin ( self , linejoin ) :
    """set to stroke linejoin .
 : param linejoin : ' undefined ' , ' miter ' , ' round ' , ' bevel '
 : type linejoin : str"""

implementation:
linejoin = getattr ( pgmagick . LineJoin , "%sJoin" % linejoin . title ( ) )
linejoin = pgmagick . DrawableStrokeLineJoin ( linejoin )
self . drawer . append ( linejoin )


signature:
def text_antialias ( self , flag = True ) :
    """text antialias
 : param flag : True or False . ( default is True )
 : type flag : bool"""

implementation:
antialias = pgmagick . DrawableTextAntialias ( flag )
self . drawer . append ( antialias )


signature:
def text_decoration ( self , decoration ) :
    """text decoration
 : param decoration : ' no ' , ' underline ' , ' overline ' , ' linethrough '
 : type decoration : str"""

implementation:
if decoration . lower ( ) == 'linethrough' :
    d = pgmagick . DecorationType . LineThroughDecoration
else :
    decoration_type_string = "%sDecoration" % decoration . title ( )
    d = getattr ( pgmagick . DecorationType , "%s" % decoration_type_string )
decoration = pgmagick . DrawableTextDecoration ( d )
self . drawer . append ( decoration )


signature:
def get_version_from_pc ( search_dirs , target ) :
    """similar to ' pkg - config - - modversion GraphicsMagick + + '"""

implementation:
for dirname in search_dirs :
    for root , dirs , files in os . walk ( dirname ) :
        for f in files :
            if f == target :
                file_path = os . path . join ( root , target )
                _tmp = _grep ( "Version: " , file_path )
                version = _tmp . split ( ) [ 1 ]
                print ( "Found version %s in file %s" % ( version , file_path ) )
                return version


signature:
def library_supports_api ( library_version , api_version , different_major_breaks_support = True ) :
    """Returns whether api _ version is supported by given library version .
 E . g . library _ version ( 1,3,21 ) returns True for api _ version ( 1,3,21 ) , ( 1,3,19 ) , ( 1,3 , ' x ' ) , ( 1,2 , ' x ' ) , ( 1 , ' x ' )
 False for ( 1,3,24 ) , ( 1,4 , ' x ' ) , ( 2 , ' x ' )
 different _ major _ breaks _ support - if enabled and library and api major versions are different always return False
 ex ) with library _ version ( 2,0,0 ) and for api _ version ( 1,3,24 ) returns False if enabled , True if disabled"""

implementation:
assert isinstance ( library_version , ( tuple , list ) )
# won ' t work with e . g . generators
assert len ( library_version ) == 3
sequence_type = type ( library_version )
# assure we will compare same types
api_version = sequence_type ( 0 if num == 'x' else num for num in api_version )
if different_major_breaks_support and library_version [ 0 ] != api_version [ 0 ] :
    return False
assert len ( api_version ) <= 3
# otherwise following comparision won ' t work as intended , e . g . ( 2 , 0 , 0 ) > ( 2 , 0 , 0 , 0)
return library_version >= api_version


signature:
def version ( ) :
    """Return version string ."""

implementation:
with io . open ( 'pgmagick/_version.py' ) as input_file :
    for line in input_file :
        if line . startswith ( '__version__' ) :
            return ast . parse ( line ) . body [ 0 ] . value . s


signature:
def post_license_request ( request ) :
    """Submission to create a license acceptance request ."""

implementation:
uuid_ = request . matchdict [ 'uuid' ]
posted_data = request . json
license_url = posted_data . get ( 'license_url' )
licensors = posted_data . get ( 'licensors' , [ ] )
with db_connect ( ) as db_conn :
    with db_conn . cursor ( ) as cursor :
        cursor . execute ( """\
SELECT l.url
FROM document_controls AS dc
LEFT JOIN licenses AS l ON (dc.licenseid = l.licenseid)
WHERE uuid = %s::UUID""" , ( uuid_ , ) )
        try : # Check that the license exists
            existing_license_url = cursor . fetchone ( ) [ 0 ]
        except TypeError : # NoneType
            if request . has_permission ( 'publish.create-identifier' ) :
                cursor . execute ( """\
INSERT INTO document_controls (uuid) VALUES (%s)""" , ( uuid_ , ) )
                existing_license_url = None
            else :
                raise httpexceptions . HTTPNotFound ( )
        if existing_license_url is None and license_url is None :
            raise httpexceptions . HTTPBadRequest ( "license_url is required" )
        elif ( license_url != existing_license_url or existing_license_url is None ) :
            cursor . execute ( """\
UPDATE document_controls AS dc
SET licenseid = l.licenseid FROM licenses AS l
WHERE url = %s and is_valid_for_publication = 't'
RETURNING dc.licenseid""" , ( license_url , ) )
            try : # Check that it is a valid license id
                cursor . fetchone ( ) [ 0 ]
            except TypeError : # None returned
                raise httpexceptions . HTTPBadRequest ( "invalid license_url" )
        upsert_license_requests ( cursor , uuid_ , licensors )
resp = request . response
resp . status_int = 202
return resp


signature:
def delete_license_request ( request ) :
    """Submission to remove a license acceptance request ."""

implementation:
uuid_ = request . matchdict [ 'uuid' ]
posted_uids = [ x [ 'uid' ] for x in request . json . get ( 'licensors' , [ ] ) ]
with db_connect ( ) as db_conn :
    with db_conn . cursor ( ) as cursor :
        remove_license_requests ( cursor , uuid_ , posted_uids )
resp = request . response
resp . status_int = 200
return resp


signature:
def get_roles_request ( request ) :
    """Returns a list of accepting roles ."""

implementation:
uuid_ = request . matchdict [ 'uuid' ]
user_id = request . matchdict . get ( 'uid' )
args = [ uuid_ ]
if user_id is not None :
    fmt_conditional = "AND user_id = %s"
    args . append ( user_id )
else :
    fmt_conditional = ""
with db_connect ( ) as db_conn :
    with db_conn . cursor ( ) as cursor :
        cursor . execute ( """\
SELECT row_to_json(combined_rows) FROM (
SELECT uuid, user_id AS uid, role_type AS role, accepted AS has_accepted
FROM role_acceptances AS la
WHERE uuid = %s {}
ORDER BY user_id ASC, role_type ASC
) as combined_rows""" . format ( fmt_conditional ) , args )
        acceptances = [ r [ 0 ] for r in cursor . fetchall ( ) ]
        if not acceptances :
            if user_id is not None :
                raise httpexceptions . HTTPNotFound ( )
            else :
                cursor . execute ( """\
SELECT TRUE FROM document_controls WHERE uuid = %s""" , ( uuid_ , ) )
                try :
                    cursor . fetchone ( ) [ 0 ]
                except TypeError : # NoneType
                    raise httpexceptions . HTTPNotFound ( )
resp_value = acceptances
if user_id is not None :
    resp_value = acceptances [ 0 ]
return resp_value


signature:
def post_roles_request ( request ) :
    """Submission to create a role acceptance request ."""

implementation:
uuid_ = request . matchdict [ 'uuid' ]
posted_roles = request . json
with db_connect ( ) as db_conn :
    with db_conn . cursor ( ) as cursor :
        cursor . execute ( """\
SELECT TRUE FROM document_controls WHERE uuid = %s::UUID""" , ( uuid_ , ) )
        try : # Check that it exists
            cursor . fetchone ( ) [ 0 ]
        except TypeError :
            if request . has_permission ( 'publish.create-identifier' ) :
                cursor . execute ( """\
INSERT INTO document_controls (uuid) VALUES (%s)""" , ( uuid_ , ) )
            else :
                raise httpexceptions . HTTPNotFound ( )
        try :
            upsert_users ( cursor , [ r [ 'uid' ] for r in posted_roles ] )
        except UserFetchError as exc :
            raise httpexceptions . HTTPBadRequest ( exc . message )
        upsert_role_requests ( cursor , uuid_ , posted_roles )
resp = request . response
resp . status_int = 202
return resp


signature:
def delete_roles_request ( request ) :
    """Submission to remove a role acceptance request ."""

implementation:
uuid_ = request . matchdict [ 'uuid' ]
posted_roles = request . json
with db_connect ( ) as db_conn :
    with db_conn . cursor ( ) as cursor :
        remove_role_requests ( cursor , uuid_ , posted_roles )
resp = request . response
resp . status_int = 200
return resp


signature:
def get_acl ( request ) :
    """Returns the ACL for the given content identified by ` ` uuid ` ` ."""

implementation:
uuid_ = request . matchdict [ 'uuid' ]
with db_connect ( ) as db_conn :
    with db_conn . cursor ( ) as cursor :
        cursor . execute ( """\
SELECT TRUE FROM document_controls WHERE uuid = %s""" , ( uuid_ , ) )
        try : # Check that it exists
            cursor . fetchone ( ) [ 0 ]
        except TypeError :
            raise httpexceptions . HTTPNotFound ( )
        cursor . execute ( """\
SELECT row_to_json(combined_rows) FROM (
SELECT uuid, user_id AS uid, permission
FROM document_acl AS acl
WHERE uuid = %s
ORDER BY user_id ASC, permission ASC
) as combined_rows""" , ( uuid_ , ) )
        acl = [ r [ 0 ] for r in cursor . fetchall ( ) ]
return acl


signature:
def post_acl_request ( request ) :
    """Submission to create an ACL ."""

implementation:
uuid_ = request . matchdict [ 'uuid' ]
posted = request . json
permissions = [ ( x [ 'uid' ] , x [ 'permission' ] , ) for x in posted ]
with db_connect ( ) as db_conn :
    with db_conn . cursor ( ) as cursor :
        cursor . execute ( """\
SELECT TRUE FROM document_controls WHERE uuid = %s::UUID""" , ( uuid_ , ) )
        try : # Check that it exists
            cursor . fetchone ( ) [ 0 ]
        except TypeError :
            if request . has_permission ( 'publish.create-identifier' ) :
                cursor . execute ( """\
INSERT INTO document_controls (uuid) VALUES (%s)""" , ( uuid_ , ) )
            else :
                raise httpexceptions . HTTPNotFound ( )
        upsert_acl ( cursor , uuid_ , permissions )
resp = request . response
resp . status_int = 202
return resp


signature:
def delete_acl_request ( request ) :
    """Submission to remove an ACL ."""

implementation:
uuid_ = request . matchdict [ 'uuid' ]
posted = request . json
permissions = [ ( x [ 'uid' ] , x [ 'permission' ] , ) for x in posted ]
with db_connect ( ) as db_conn :
    with db_conn . cursor ( ) as cursor :
        remove_acl ( cursor , uuid_ , permissions )
resp = request . response
resp . status_int = 200
return resp


signature:
def processor ( ) : # pragma : no cover
    """Churns over PostgreSQL notifications on configured channels .
 This requires the application be setup and the registry be available .
 This function uses the database connection string and a list of
 pre configured channels ."""

implementation:
registry = get_current_registry ( )
settings = registry . settings
connection_string = settings [ CONNECTION_STRING ]
channels = _get_channels ( settings )
# Code adapted from
# http : / / initd . org / psycopg / docs / advanced . html # asynchronous - notifications
with psycopg2 . connect ( connection_string ) as conn :
    conn . set_isolation_level ( ISOLATION_LEVEL_AUTOCOMMIT )
    with conn . cursor ( ) as cursor :
        for channel in channels :
            cursor . execute ( 'LISTEN {}' . format ( channel ) )
            logger . debug ( 'Waiting for notifications on channel "{}"' . format ( channel ) )
    registry . notify ( ChannelProcessingStartUpEvent ( ) )
    rlist = [ conn ]
    # wait until ready for reading
    wlist = [ ]
    # wait until ready for writing
    xlist = [ ]
    # wait for an " exceptional condition "
    timeout = 5
    while True :
        if select . select ( rlist , wlist , xlist , timeout ) != ( [ ] , [ ] , [ ] ) :
            conn . poll ( )
            while conn . notifies :
                notif = conn . notifies . pop ( 0 )
                logger . debug ( 'Got NOTIFY: pid={} channel={} payload={}' . format ( notif . pid , notif . channel , notif . payload ) )
                event = create_pg_notify_event ( notif )
                try :
                    registry . notify ( event )
                except Exception :
                    logger . exception ( 'Logging an uncaught exception' )


signature:
def lookup_api_key_info ( ) :
    """Given a dbapi cursor , lookup all the api keys and their information ."""

implementation:
info = { }
with db_connect ( ) as conn :
    with conn . cursor ( ) as cursor :
        cursor . execute ( ALL_KEY_INFO_SQL_STMT )
        for row in cursor . fetchall ( ) :
            id , key , name , groups = row
            user_id = "api_key:{}" . format ( id )
            info [ key ] = dict ( id = id , user_id = user_id , name = name , groups = groups )
return info


signature:
def includeme ( config ) :
    """Configuration include fuction for this module"""

implementation:
api_key_authn_policy = APIKeyAuthenticationPolicy ( )
config . include ( 'openstax_accounts' )
openstax_authn_policy = config . registry . getUtility ( IOpenstaxAccountsAuthenticationPolicy )
# Set up api & user authentication policies .
policies = [ api_key_authn_policy , openstax_authn_policy ]
authn_policy = MultiAuthenticationPolicy ( policies )
config . set_authentication_policy ( authn_policy )
# Set up the authorization policy .
authz_policy = ACLAuthorizationPolicy ( )
config . set_authorization_policy ( authz_policy )


signature:
def expandvars_dict ( settings ) :
    """Expands all environment variables in a settings dictionary ."""

implementation: return dict ( ( key , os . path . expandvars ( value ) ) for key , value in settings . iteritems ( ) )


signature:
def initialize_sentry_integration ( ) : # pragma : no cover
    """Used to optionally initialize the Sentry service with this app .
 See https : / / docs . sentry . io / platforms / python / pyramid /"""

implementation:
# This function is not under coverage because it is boilerplate
# from the Sentry documentation .
try :
    import sentry_sdk
    from sentry_sdk . integrations . pyramid import PyramidIntegration
    from sentry_sdk . integrations . celery import CeleryIntegration
except ImportError :
    warnings . warn ( "Sentry is not configured because the Sentry SDK " "(sentry_sdk package) is not installed" , UserWarning , )
    return
# bail out early
try :
    dsn = os . environ [ 'SENTRY_DSN' ]
except KeyError :
    warnings . warn ( "Sentry is not configured because SENTRY_DSN " "was not supplied." , UserWarning , )
else :
    sentry_sdk . init ( dsn = dsn , integrations = [ PyramidIntegration ( ) , CeleryIntegration ( ) ] , )


signature:
def task ( ** kwargs ) :
    """A function task decorator used in place of ` ` @ celery _ app . task ` ` ."""

implementation:
def wrapper ( wrapped ) :
    def callback ( scanner , name , obj ) :
        celery_app = scanner . config . registry . celery_app
        celery_app . task ( ** kwargs ) ( obj )
    venusian . attach ( wrapped , callback )
    return wrapped
return wrapper


signature:
def _make_celery_app ( config ) :
    """This exposes the celery app . The app is actually created as part
 of the configuration . However , this does make the celery app functional
 as a stand - alone celery application .
 This puts the pyramid configuration object on the celery app to be
 used for making the registry available to tasks running inside the
 celery worker process pool . See ` ` CustomTask . _ _ call _ _ ` ` ."""

implementation:
# Tack the pyramid config on the celery app for later use .
config . registry . celery_app . conf [ 'pyramid_config' ] = config
return config . registry . celery_app


signature:
def post_publication_processing ( event , cursor ) :
    """Process post - publication events coming out of the database ."""

implementation:
module_ident , ident_hash = event . module_ident , event . ident_hash
celery_app = get_current_registry ( ) . celery_app
# Check baking is not already queued .
cursor . execute ( 'SELECT result_id::text ' 'FROM document_baking_result_associations ' 'WHERE module_ident = %s' , ( module_ident , ) )
for result in cursor . fetchall ( ) :
    state = celery_app . AsyncResult ( result [ 0 ] ) . state
    if state in ( 'QUEUED' , 'STARTED' , 'RETRY' ) :
        logger . debug ( 'Already queued module_ident={} ident_hash={}' . format ( module_ident , ident_hash ) )
        return
logger . debug ( 'Queued for processing module_ident={} ident_hash={}' . format ( module_ident , ident_hash ) )
recipe_ids = _get_recipe_ids ( module_ident , cursor )
update_module_state ( cursor , module_ident , 'processing' , recipe_ids [ 0 ] )
# Commit the state change before preceding .
cursor . connection . commit ( )
# Start of task
# FIXME Looking up the task isn ' t the most clear usage here .
task_name = 'cnxpublishing.subscribers.baking_processor'
baking_processor = celery_app . tasks [ task_name ]
result = baking_processor . delay ( module_ident , ident_hash )
baking_processor . backend . store_result ( result . id , None , 'QUEUED' )
# Save the mapping between a celery task and this event .
track_baking_proc_state ( result , module_ident , cursor )


signature:
def parse_archive_uri ( uri ) :
    """Given an archive URI , parse to a split ident - hash ."""

implementation:
parsed = urlparse ( uri )
path = parsed . path . rstrip ( '/' ) . split ( '/' )
ident_hash = path [ - 1 ]
ident_hash = unquote ( ident_hash )
return ident_hash


signature:
def declare_api_routes ( config ) :
    """Declaration of routing"""

implementation:
add_route = config . add_route
add_route ( 'get-content' , '/contents/{ident_hash}' )
add_route ( 'get-resource' , '/resources/{hash}' )
# User actions API
add_route ( 'license-request' , '/contents/{uuid}/licensors' )
add_route ( 'roles-request' , '/contents/{uuid}/roles' )
add_route ( 'acl-request' , '/contents/{uuid}/permissions' )
# Publishing API
add_route ( 'publications' , '/publications' )
add_route ( 'get-publication' , '/publications/{id}' )
add_route ( 'publication-license-acceptance' , '/publications/{id}/license-acceptances/{uid}' )
add_route ( 'publication-role-acceptance' , '/publications/{id}/role-acceptances/{uid}' )
# TODO ( 8 - May - 12017 ) Remove because the term collate is being phased out .
add_route ( 'collate-content' , '/contents/{ident_hash}/collate-content' )
add_route ( 'bake-content' , '/contents/{ident_hash}/baked' )
# Moderation routes
add_route ( 'moderation' , '/moderations' )
add_route ( 'moderate' , '/moderations/{id}' )
add_route ( 'moderation-rss' , '/feeds/moderations.rss' )
# API Key routes
add_route ( 'api-keys' , '/api-keys' )
add_route ( 'api-key' , '/api-keys/{id}' )


signature:
def declare_browsable_routes ( config ) :
    """Declaration of routes that can be browsed by users ."""

implementation:
# This makes our routes slashed , which is good browser behavior .
config . add_notfound_view ( default_exceptionresponse_view , append_slash = True )
add_route = config . add_route
add_route ( 'admin-index' , '/a/' )
add_route ( 'admin-moderation' , '/a/moderation/' )
add_route ( 'admin-api-keys' , '/a/api-keys/' )
add_route ( 'admin-add-site-messages' , '/a/site-messages/' , request_method = 'GET' )
add_route ( 'admin-add-site-messages-POST' , '/a/site-messages/' , request_method = 'POST' )
add_route ( 'admin-delete-site-messages' , '/a/site-messages/' , request_method = 'DELETE' )
add_route ( 'admin-edit-site-message' , '/a/site-messages/{id}/' , request_method = 'GET' )
add_route ( 'admin-edit-site-message-POST' , '/a/site-messages/{id}/' , request_method = 'POST' )
add_route ( 'admin-content-status' , '/a/content-status/' )
add_route ( 'admin-content-status-single' , '/a/content-status/{uuid}' )
add_route ( 'admin-print-style' , '/a/print-style/' )
add_route ( 'admin-print-style-single' , '/a/print-style/{style}' )


signature:
def includeme ( config ) :
    """Declare all routes ."""

implementation:
config . include ( 'pyramid_jinja2' )
config . add_jinja2_renderer ( '.html' )
config . add_jinja2_renderer ( '.rss' )
config . add_static_view ( name = '/a/static' , path = "cnxpublishing:static/" )
# Commit the configuration otherwise the jija2 _ env won ' t have
# a ` globals ` assignment .
config . commit ( )
# Place a few globals in the template environment .
from cnxdb . ident_hash import join_ident_hash
for ext in ( '.html' , '.rss' , ) :
    jinja2_env = config . get_jinja2_environment ( ext )
    jinja2_env . globals . update ( join_ident_hash = join_ident_hash , )
declare_api_routes ( config )
declare_browsable_routes ( config )


signature:
def _formatter_callback_factory ( ) : # pragma : no cover
    """Returns a list of includes to be given to ` cnxepub . collation . collate ` ."""

implementation:
includes = [ ]
exercise_url_template = '{baseUrl}/api/exercises?q={field}:"{{itemCode}}"'
settings = get_current_registry ( ) . settings
exercise_base_url = settings . get ( 'embeddables.exercise.base_url' , None )
exercise_matches = [ match . split ( ',' , 1 ) for match in aslist ( settings . get ( 'embeddables.exercise.match' , '' ) , flatten = False ) ]
exercise_token = settings . get ( 'embeddables.exercise.token' , None )
mathml_url = settings . get ( 'mathmlcloud.url' , None )
memcache_servers = settings . get ( 'memcache_servers' )
if memcache_servers :
    memcache_servers = memcache_servers . split ( )
else :
    memcache_servers = None
if exercise_base_url and exercise_matches :
    mc_client = None
    if memcache_servers :
        mc_client = memcache . Client ( memcache_servers , debug = 0 )
    for ( exercise_match , exercise_field ) in exercise_matches :
        template = exercise_url_template . format ( baseUrl = exercise_base_url , field = exercise_field )
        includes . append ( exercise_callback_factory ( exercise_match , template , mc_client , exercise_token , mathml_url ) )
return includes


signature:
def bake ( binder , recipe_id , publisher , message , cursor ) :
    """Given a ` Binder ` as ` binder ` , bake the contents and
 persist those changes alongside the published content ."""

implementation:
recipe = _get_recipe ( recipe_id , cursor )
includes = _formatter_callback_factory ( )
binder = collate_models ( binder , ruleset = recipe , includes = includes )
def flatten_filter ( model ) :
    return ( isinstance ( model , cnxepub . CompositeDocument ) or ( isinstance ( model , cnxepub . Binder ) and model . metadata . get ( 'type' ) == 'composite-chapter' ) )
def only_documents_filter ( model ) :
    return isinstance ( model , cnxepub . Document ) and not isinstance ( model , cnxepub . CompositeDocument )
for doc in cnxepub . flatten_to ( binder , flatten_filter ) :
    publish_composite_model ( cursor , doc , binder , publisher , message )
for doc in cnxepub . flatten_to ( binder , only_documents_filter ) :
    publish_collated_document ( cursor , doc , binder )
tree = cnxepub . model_to_tree ( binder )
publish_collated_tree ( cursor , tree )
return [ ]


signature:
def volcano ( differential_dfs , title = 'Axial Volcano Plot' , scripts_mode = "CDN" , data_mode = "directory" , organism = "human" , q_value_column_name = "q" , log2FC_column_name = "logFC" , output_dir = "." , filename = "volcano.html" , version = this_version ) :
    """Arguments :
 differential _ dfs ( dict or pandas . DataFrame ) : python dict of names to pandas dataframes , or a single dataframe , indexed by gene symbols which must have columns named log2FC and qval .
 title ( str ) : The title of the plot ( to be embedded in the html ) .
 scripts _ mode ( str ) : Choose from [ ` " CDN " ` , ` " directory " ` , ` " inline " ` ] :
 - ` " CDN " ` compiles a single HTML page with links to scripts hosted on a CDN ,
 - ` " directory " ` compiles a directory with all scripts locally cached ,
 - ` " inline " ` compiles a single HTML file with all scripts / styles inlined .
 data _ mode ( str ) : Choose from [ " directory " , " inline " ] :
 - " directory " compiles a directory with all data locally cached ,
 - " inline " compiles a single HTML file with all data inlined .
 organism ( str ) : ` " human " ` or ` " mouse " `
 q _ value _ column _ name ( str ) :
 log2FC _ column _ name ( str ) :
 output _ dir ( str ) : the directory in which to output the file
 filename ( str ) : the filename of the output file
 version ( str ) : the version of the javascripts to use .
 Leave the default to pin the version , or choose " latest " to get updates ,
 or choose part of the version string to get minor updates .
 Returns :
 Path : The filepath which the html was outputted to ."""

implementation:
output_dir = Path ( output_dir )
output_dir . mkdir ( exist_ok = True , parents = True )
# Data = = = = =
if isinstance ( differential_dfs , pd . DataFrame ) :
    differential_dfs = { 'differential' : differential_dfs }
for name , df in differential_dfs . items ( ) :
    df = df [ [ q_value_column_name , log2FC_column_name ] ]
    df . columns = [ 'q' , 'logFC' ]
    df = df . round ( 2 )
    # TODO drop all zero rows
    _verify_differential_df ( df )
    del differential_dfs [ name ]
    differential_dfs [ _sanitize ( name ) ] = df
names_and_differentials = f"var names_and_differentials = { '{'+ ','.join([_quote(name)+': '+df.to_json(orient='index') for name, df in differential_dfs.items()]) +'}' };"
data_block = _data_block ( data_mode , [ ( 'names_and_differentials' , names_and_differentials ) ] , output_dir , include_gene_sets = False , organism = organism )
# Scripts = = = = =
scripts = third_party_scripts + [ CDN_url ( version ) + "js/util.js" , CDN_url ( version ) + "js/GOrilla.js" , CDN_url ( version ) + "js/volcano.js" ]
scripts_block = _scripts_block ( scripts , scripts_mode , output_dir )
html = templateEnv . get_template ( 'volcano.html.j2' ) . render ( title = title , scripts_block = scripts_block + '\n' + data_block , organism = "HOMO_SAPIENS" )
( output_dir / filename ) . write_text ( html )
return ( output_dir / filename ) . resolve ( )


signature:
def heatmap ( genes_by_samples_matrix , sample_attributes , title = 'Axial Heatmap' , scripts_mode = "CDN" , data_mode = "directory" , organism = "human" , separate_zscore_by = [ "system" ] , output_dir = "." , filename = "heatmap.html" , version = this_version ) :
    """Arguments :
 genes _ by _ samples _ matrix ( pandas . DataFrame ) : dataframe indexed by genes , columns are samples
 sample _ attributes ( pandas . DataFrame ) : dataframe indexed by samples , columns are sample attributes ( e . g . classes )
 title ( str ) : The title of the plot ( to be embedded in the html ) .
 scripts _ mode ( str ) : Choose from [ ` " CDN " ` , ` " directory " ` , ` " inline " ` ] :
 - ` " CDN " ` compiles a single HTML page with links to scripts hosted on a CDN ,
 - ` " directory " ` compiles a directory with all scripts locally cached ,
 - ` " inline " ` compiles a single HTML file with all scripts / styles inlined .
 data _ mode ( str ) : Choose from [ " directory " , " inline " ] :
 - " directory " compiles a directory with all data locally cached ,
 - " inline " compiles a single HTML file with all data inlined .
 organism ( str ) : ` " human " ` or ` " mouse " `
 separate _ zscore _ by ( list ) :
 output _ dir ( str ) : the directory in which to output the file
 filename ( str ) : the filename of the output file
 version ( str ) : the version of the javascripts to use .
 Leave the default to pin the version , or choose " latest " to get updates ,
 or choose part of the version string to get minor updates .
 Returns :
 Path : The filepath which the html was outputted to ."""

implementation:
output_dir = Path ( output_dir )
output_dir . mkdir ( exist_ok = True , parents = True )
# Data = = = = =
_verify_sample_by_genes_matrix ( genes_by_samples_matrix )
_verify_sample_attributes ( genes_by_samples_matrix , sample_attributes )
genes_by_samples_matrix = genes_by_samples_matrix . round ( 2 )
# TODO drop all zero rows
matrix = f"var matrix = {genes_by_samples_matrix.to_json(orient='columns')};"
classes = f"var classes = {sample_attributes.to_json(orient='index')};"
data_block = _data_block ( data_mode , [ ( 'matrix' , matrix ) , ( 'classes' , classes ) ] , output_dir , organism = organism )
# Scripts = = = = =
scripts = third_party_scripts + [ CDN_url ( version ) + "js/util.js" , CDN_url ( version ) + "js/reorder.js" , CDN_url ( version ) + "js/heatmap.js" ]
scripts_block = _scripts_block ( scripts , scripts_mode , output_dir )
html = templateEnv . get_template ( 'heatmap.html.j2' ) . render ( title = title , scripts_block = scripts_block + '\n' + data_block , separate_zscore_by = separate_zscore_by )
( output_dir / filename ) . write_text ( html )
return ( output_dir / filename ) . resolve ( )


signature:
def graph ( networkx_graph , title = 'Axial Graph Visualization' , scripts_mode = "CDN" , data_mode = "directory" , output_dir = "." , filename = "graph.html" , version = this_version ) :
    """Arguments :
 networkx _ graph ( networkx . Graph ) : any instance of networkx . Graph
 title ( str ) : The title of the plot ( to be embedded in the html ) .
 scripts _ mode ( str ) : Choose from [ ` " CDN " ` , ` " directory " ` , ` " inline " ` ] :
 - ` " CDN " ` compiles a single HTML page with links to scripts hosted on a CDN ,
 - ` " directory " ` compiles a directory with all scripts locally cached ,
 - ` " inline " ` compiles a single HTML file with all scripts / styles inlined .
 data _ mode ( str ) : Choose from [ " directory " , " inline " ] :
 - " directory " compiles a directory with all data locally cached ,
 - " inline " compiles a single HTML file with all data inlined .
 output _ dir ( str ) : the directory in which to output the file
 filename ( str ) : the filename of the output file
 version ( str ) : the version of the javascripts to use .
 Leave the default to pin the version , or choose " latest " to get updates ,
 or choose part of the version string to get minor updates .
 Returns :
 Path : The filepath which the html was outputted to ."""

implementation:
output_dir = Path ( output_dir )
output_dir . mkdir ( exist_ok = True , parents = True )
# Scripts = = = = =
scripts = third_party_scripts + [ CDN_url ( version ) + "js/cola.min.js" , CDN_url ( version ) + "js/graph.js" ]
scripts_block = _scripts_block ( scripts , scripts_mode , output_dir )
# Data = = = = =
graph_json = nx_json . node_link_data ( networkx_graph )
for node in graph_json [ 'nodes' ] :
    for attr , val in node . items ( ) :
        if isinstance ( val , numbers . Number ) :
            node [ attr ] = round ( val , 2 )
for link in graph_json [ 'links' ] :
    for attr , val in link . items ( ) :
        if isinstance ( val , numbers . Number ) :
            link [ attr ] = round ( val , 2 )
graph_json = f"var graph = {json.dumps(graph_json)};"
data_block = _data_block ( data_mode , [ ( 'graph' , graph_json ) ] , output_dir )
html = templateEnv . get_template ( 'graph.html.j2' ) . render ( title = title , scripts_block = scripts_block + '\n' + data_block , nodes = networkx_graph . nodes ( ) )
( output_dir / filename ) . write_text ( html )
return ( output_dir / filename ) . resolve ( )


signature:
def db_connect ( connection_string = None , ** kwargs ) :
    """Function to supply a database connection object ."""

implementation:
if connection_string is None :
    connection_string = get_current_registry ( ) . settings [ CONNECTION_STRING ]
db_conn = psycopg2 . connect ( connection_string , ** kwargs )
try :
    with db_conn :
        yield db_conn
finally :
    db_conn . close ( )


signature:
def with_db_cursor ( func ) :
    """Decorator that supplies a cursor to the function .
 This passes in a psycopg2 Cursor as the argument ' cursor ' .
 It also accepts a cursor if one is given ."""

implementation:
@ functools . wraps ( func )
def wrapped ( * args , ** kwargs ) :
    if 'cursor' in kwargs or func . func_code . co_argcount == len ( args ) :
        return func ( * args , ** kwargs )
    with db_connect ( ) as db_connection :
        with db_connection . cursor ( ) as cursor :
            kwargs [ 'cursor' ] = cursor
            return func ( * args , ** kwargs )
return wrapped


signature:
def _role_type_to_db_type ( type_ ) :
    """Translates a role type ( a value found in
 ` ` cnxepub . ATTRIBUTED _ ROLE _ KEYS ` ` ) to a database compatible
 value for ` ` role _ types ` ` ."""

implementation:
with db_connect ( ) as db_conn :
    with db_conn . cursor ( ) as cursor :
        cursor . execute ( """\
WITH unnested_role_types AS (
  SELECT unnest(enum_range(NULL::role_types)) as role_type
  ORDER BY role_type ASC)
SELECT array_agg(role_type)::text[] FROM unnested_role_types""" )
        db_types = cursor . fetchone ( ) [ 0 ]
return dict ( zip ( cnxepub . ATTRIBUTED_ROLE_KEYS , db_types ) ) [ type_ ]


signature:
def _dissect_roles ( metadata ) :
    """Given a model ' s ` ` metadata ` ` , iterate over the roles .
 Return values are the role identifier and role type as a tuple ."""

implementation:
for role_key in cnxepub . ATTRIBUTED_ROLE_KEYS :
    for user in metadata . get ( role_key , [ ] ) :
        if user [ 'type' ] != 'cnx-id' :
            raise ValueError ( "Archive only accepts Connexions users." )
        uid = parse_user_uri ( user [ 'id' ] )
        yield uid , role_key
raise StopIteration ( )


signature:
def upsert_pending_licensors ( cursor , document_id ) :
    """Update or insert records for pending license acceptors ."""

implementation:
cursor . execute ( """\
SELECT "uuid", "metadata"
FROM pending_documents
WHERE id = %s""" , ( document_id , ) )
uuid_ , metadata = cursor . fetchone ( )
acceptors = set ( [ uid for uid , type_ in _dissect_roles ( metadata ) ] )
# Acquire a list of existing acceptors .
cursor . execute ( """\
SELECT "user_id", "accepted"
FROM license_acceptances
WHERE uuid = %s""" , ( uuid_ , ) )
existing_acceptors_mapping = dict ( cursor . fetchall ( ) )
# Who ' s not in the existing list ?
existing_acceptors = set ( existing_acceptors_mapping . keys ( ) )
new_acceptors = acceptors . difference ( existing_acceptors )
# Insert the new licensor acceptors .
for acceptor in new_acceptors :
    cursor . execute ( """\
INSERT INTO license_acceptances
  ("uuid", "user_id", "accepted")
VALUES (%s, %s, NULL)""" , ( uuid_ , acceptor , ) )
# Has everyone already accepted ?
cursor . execute ( """\
SELECT user_id
FROM license_acceptances
WHERE
  uuid = %s
  AND
  (accepted is UNKNOWN OR accepted is FALSE)""" , ( uuid_ , ) )
defectors = set ( cursor . fetchall ( ) )
if not defectors : # Update the pending document license acceptance state .
    cursor . execute ( """\
update pending_documents set license_accepted = 't'
where id = %s""" , ( document_id , ) )


signature:
def upsert_pending_roles ( cursor , document_id ) :
    """Update or insert records for pending document role acceptance ."""

implementation:
cursor . execute ( """\
SELECT "uuid", "metadata"
FROM pending_documents
WHERE id = %s""" , ( document_id , ) )
uuid_ , metadata = cursor . fetchone ( )
acceptors = set ( [ ( uid , _role_type_to_db_type ( type_ ) , ) for uid , type_ in _dissect_roles ( metadata ) ] )
# Upsert the user info .
upsert_users ( cursor , [ x [ 0 ] for x in acceptors ] )
# Acquire a list of existing acceptors .
cursor . execute ( """\
SELECT user_id, role_type
FROM role_acceptances
WHERE uuid = %s""" , ( uuid_ , ) )
existing_roles = set ( [ ( r , t , ) for r , t in cursor . fetchall ( ) ] )
# Who ' s not in the existing list ?
existing_acceptors = existing_roles
new_acceptors = acceptors . difference ( existing_acceptors )
# Insert the new role acceptors .
for acceptor , type_ in new_acceptors :
    cursor . execute ( """\
INSERT INTO role_acceptances
  ("uuid", "user_id", "role_type", "accepted")
        VALUES (%s, %s, %s, DEFAULT)""" , ( uuid_ , acceptor , type_ ) )
# Has everyone already accepted ?
cursor . execute ( """\
SELECT user_id
FROM role_acceptances
WHERE
  uuid = %s
  AND
  (accepted is UNKNOWN OR accepted is FALSE)""" , ( uuid_ , ) )
defectors = set ( cursor . fetchall ( ) )
if not defectors : # Update the pending document license acceptance state .
    cursor . execute ( """\
update pending_documents set roles_accepted = 't'
where id = %s""" , ( document_id , ) )


signature:
def obtain_licenses ( ) :
    """Obtain the licenses in a dictionary form , keyed by url ."""

implementation:
with db_connect ( ) as db_conn :
    with db_conn . cursor ( ) as cursor :
        cursor . execute ( """\
SELECT combined_row.url, row_to_json(combined_row) FROM (
  SELECT "code", "version", "name", "url", "is_valid_for_publication"
  FROM licenses) AS combined_row""" )
        licenses = { r [ 0 ] : r [ 1 ] for r in cursor . fetchall ( ) }
return licenses


signature:
def _validate_license ( model ) :
    """Given the model , check the license is one valid for publication ."""

implementation:
license_mapping = obtain_licenses ( )
try :
    license_url = model . metadata [ 'license_url' ]
except KeyError :
    raise exceptions . MissingRequiredMetadata ( 'license_url' )
try :
    license = license_mapping [ license_url ]
except KeyError :
    raise exceptions . InvalidLicense ( license_url )
if not license [ 'is_valid_for_publication' ] :
    raise exceptions . InvalidLicense ( license_url )


signature:
def _validate_roles ( model ) :
    """Given the model , check that all the metadata role values
 have valid information in them and any required metadata fields
 contain values ."""

implementation:
required_roles = ( ATTRIBUTED_ROLE_KEYS [ 0 ] , ATTRIBUTED_ROLE_KEYS [ 4 ] , )
for role_key in ATTRIBUTED_ROLE_KEYS :
    try :
        roles = model . metadata [ role_key ]
    except KeyError :
        if role_key in required_roles :
            raise exceptions . MissingRequiredMetadata ( role_key )
    else :
        if role_key in required_roles and len ( roles ) == 0 :
            raise exceptions . MissingRequiredMetadata ( role_key )
    for role in roles :
        if role . get ( 'type' ) != 'cnx-id' :
            raise exceptions . InvalidRole ( role_key , role )


signature:
def _validate_derived_from ( cursor , model ) :
    """Given a database cursor and model , check the derived - from
 value accurately points to content in the archive .
 The value can be nothing or must point to existing content ."""

implementation:
derived_from_uri = model . metadata . get ( 'derived_from_uri' )
if derived_from_uri is None :
    return
# bail out early
# Can we parse the value ?
try :
    ident_hash = parse_archive_uri ( derived_from_uri )
    uuid_ , version = split_ident_hash ( ident_hash , split_version = True )
except ( ValueError , IdentHashSyntaxError , IdentHashShortId ) as exc :
    raise exceptions . InvalidMetadata ( 'derived_from_uri' , derived_from_uri , original_exception = exc )
# Is the ident - hash a valid pointer ?
args = [ uuid_ ]
table = 'modules'
version_condition = ''
if version != ( None , None , ) :
    args . extend ( version )
    table = 'modules'
    version_condition = " AND major_version = %s" " AND minor_version {} %s" . format ( version [ 1 ] is None and 'is' or '=' )
cursor . execute ( """SELECT 't' FROM {} WHERE uuid::text = %s{}""" . format ( table , version_condition ) , args )
try :
    _exists = cursor . fetchone ( ) [ 0 ]
    # noqa
except TypeError : # None type
    raise exceptions . InvalidMetadata ( 'derived_from_uri' , derived_from_uri )
# Assign the derived _ from value so that we don ' t have to split it again .
model . metadata [ 'derived_from' ] = ident_hash


signature:
def _validate_subjects ( cursor , model ) :
    """Give a database cursor and model , check the subjects against
 the subject vocabulary ."""

implementation:
subject_vocab = [ term [ 0 ] for term in acquire_subject_vocabulary ( cursor ) ]
subjects = model . metadata . get ( 'subjects' , [ ] )
invalid_subjects = [ s for s in subjects if s not in subject_vocab ]
if invalid_subjects :
    raise exceptions . InvalidMetadata ( 'subjects' , invalid_subjects )


signature:
def validate_model ( cursor , model ) :
    """Validates the model using a series of checks on bits of the data ."""

implementation:
# Check the license is one valid for publication .
_validate_license ( model )
_validate_roles ( model )
# Other required metadata includes : title , summary
required_metadata = ( 'title' , 'summary' , )
for metadata_key in required_metadata :
    if model . metadata . get ( metadata_key ) in [ None , '' , [ ] ] :
        raise exceptions . MissingRequiredMetadata ( metadata_key )
# Ensure that derived - from values are either None
# or point at a live record in the archive .
_validate_derived_from ( cursor , model )
# FIXME Valid language code ?
# Are the given ' subjects '
_validate_subjects ( cursor , model )


signature:
def is_publication_permissible ( cursor , publication_id , uuid_ ) :
    """Check the given publisher of this publication given
 by ` ` publication _ id ` ` is allowed to publish the content given
 by ` ` uuid ` ` ."""

implementation:
# Check the publishing user has permission to publish
cursor . execute ( """\
SELECT 't'::boolean
FROM
  pending_documents AS pd
  NATURAL JOIN document_acl AS acl
  JOIN publications AS p ON (pd.publication_id = p.id)
WHERE
  p.id = %s
  AND
  pd.uuid = %s
  AND
  p.publisher = acl.user_id
  AND
  acl.permission = 'publish'""" , ( publication_id , uuid_ , ) )
try :
    is_allowed = cursor . fetchone ( ) [ 0 ]
except TypeError :
    is_allowed = False
return is_allowed


signature:
def add_pending_model ( cursor , publication_id , model ) :
    """Adds a model ( binder or document ) that is awaiting publication
 to the database ."""

implementation:
# FIXME Too much happening here . . .
assert isinstance ( model , ( cnxepub . Document , cnxepub . Binder , ) ) , type ( model )
uri = model . get_uri ( 'cnx-archive' )
if uri is not None :
    ident_hash = parse_archive_uri ( uri )
    id , version = split_ident_hash ( ident_hash , split_version = True )
    cursor . execute ( """\
SELECT major_version + 1 as next_version
FROM latest_modules
WHERE uuid = %s
UNION ALL
SELECT 1 as next_version
ORDER BY next_version DESC
LIMIT 1
""" , ( id , ) )
    next_major_version = cursor . fetchone ( ) [ 0 ]
    if isinstance ( model , cnxepub . Document ) :
        version = ( next_major_version , None , )
    else : # . . . assume it ' s a binder .
        version = ( next_major_version , 1 , )
else :
    cursor . execute ( """\
WITH
control_insert AS (
  INSERT INTO document_controls (uuid) VALUES (DEFAULT) RETURNING uuid),
acl_insert AS (
  INSERT INTO document_acl (uuid, user_id, permission)
  VALUES ((SELECT uuid FROM control_insert),
          (SELECT publisher FROM publications WHERE id = %s),
          'publish'::permission_type))
SELECT uuid FROM control_insert""" , ( publication_id , ) )
    id = cursor . fetchone ( ) [ 0 ]
    if isinstance ( model , cnxepub . Document ) :
        version = ( 1 , None , )
    else : # . . . assume it ' s a binder .
        version = ( 1 , 1 , )
type_ = _get_type_name ( model )
model . id = str ( id )
model . metadata [ 'version' ] = '.' . join ( [ str ( v ) for v in version if v ] )
args = [ publication_id , id , version [ 0 ] , version [ 1 ] , type_ , json . dumps ( model . metadata ) ]
cursor . execute ( """\
INSERT INTO "pending_documents"
  ("publication_id", "uuid", "major_version", "minor_version", "type",
    "license_accepted", "roles_accepted", "metadata")
VALUES (%s, %s, %s, %s, %s, 'f', 'f', %s)
RETURNING "id", "uuid", module_version("major_version", "minor_version")
""" , args )
pending_id , uuid_ , version = cursor . fetchone ( )
pending_ident_hash = join_ident_hash ( uuid_ , version )
# Assign the new ident - hash to the document for later use .
request = get_current_request ( )
path = request . route_path ( 'get-content' , ident_hash = pending_ident_hash )
model . set_uri ( 'cnx-archive' , path )
# Check if the publication is allowed for the publishing user .
if not is_publication_permissible ( cursor , publication_id , id ) : # Set the failure but continue the operation of inserting
# the pending document .
    exc = exceptions . NotAllowed ( id )
    exc . publication_id = publication_id
    exc . pending_document_id = pending_id
    exc . pending_ident_hash = pending_ident_hash
    set_publication_failure ( cursor , exc )
try :
    validate_model ( cursor , model )
except exceptions . PublicationException as exc :
    exc_info = sys . exc_info ( )
    exc . publication_id = publication_id
    exc . pending_document_id = pending_id
    exc . pending_ident_hash = pending_ident_hash
    try :
        set_publication_failure ( cursor , exc )
    except BaseException :
        import traceback
        print ( "Critical data error. Immediate attention is " "required. On publication at '{}'." . format ( publication_id ) , file = sys . stderr )
        # Print the critical exception .
        traceback . print_exc ( )
        # Raise the previous exception , so we know the original cause .
        raise exc_info [ 0 ] , exc_info [ 1 ] , exc_info [ 2 ]
else :
    upsert_pending_licensors ( cursor , pending_id )
    upsert_pending_roles ( cursor , pending_id )
    notify_users ( cursor , pending_id )
return pending_ident_hash


signature:
def lookup_document_pointer ( ident_hash , cursor ) :
    """Lookup a document by id and version ."""

implementation:
id , version = split_ident_hash ( ident_hash , split_version = True )
stmt = "SELECT name FROM modules WHERE uuid = %s"
args = [ id ]
if version and version [ 0 ] is not None :
    operator = version [ 1 ] is None and 'is' or '='
    stmt += " AND (major_version = %s AND minor_version {} %s)" . format ( operator )
    args . extend ( version )
cursor . execute ( stmt , args )
try :
    title = cursor . fetchone ( ) [ 0 ]
except TypeError :
    raise DocumentLookupError ( )
else :
    metadata = { 'title' : title }
return cnxepub . DocumentPointer ( ident_hash , metadata )


signature:
def set_publication_failure ( cursor , exc ) :
    """Given a publication exception , set the publication as failed and
 append the failure message to the publication record ."""

implementation:
publication_id = exc . publication_id
if publication_id is None :
    raise ValueError ( "Exception must have a ``publication_id`` value." )
cursor . execute ( """\
SELECT "state_messages"
FROM publications
WHERE id = %s""" , ( publication_id , ) )
state_messages = cursor . fetchone ( ) [ 0 ]
if state_messages is None :
    state_messages = [ ]
entry = exc . __dict__
entry [ 'message' ] = exc . message
state_messages . append ( entry )
state_messages = json . dumps ( state_messages )
cursor . execute ( """\
UPDATE publications SET ("state", "state_messages") = (%s, %s)
WHERE id = %s""" , ( 'Failed/Error' , state_messages , publication_id , ) )


signature:
def add_publication ( cursor , epub , epub_file , is_pre_publication = False ) :
    """Adds a publication entry and makes each item
 a pending document ."""

implementation:
publisher = epub [ 0 ] . metadata [ 'publisher' ]
publish_message = epub [ 0 ] . metadata [ 'publication_message' ]
epub_binary = psycopg2 . Binary ( epub_file . read ( ) )
args = ( publisher , publish_message , epub_binary , is_pre_publication , )
cursor . execute ( """\
INSERT INTO publications
  ("publisher", "publication_message", "epub", "is_pre_publication")
VALUES (%s, %s, %s, %s)
RETURNING id
""" , args )
publication_id = cursor . fetchone ( ) [ 0 ]
insert_mapping = { }
models = set ( [ ] )
for package in epub :
    binder = cnxepub . adapt_package ( package )
    if binder in models :
        continue
    for document in cnxepub . flatten_to_documents ( binder ) :
        if document not in models :
            ident_hash = add_pending_model ( cursor , publication_id , document )
            insert_mapping [ document . id ] = ident_hash
            models . add ( document )
    # The binding object could be translucent / see - through ,
    # ( case for a binder that only contains loose - documents ) .
    # Otherwise we should also publish the the binder .
    if not binder . is_translucent :
        ident_hash = add_pending_model ( cursor , publication_id , binder )
        insert_mapping [ binder . id ] = ident_hash
        models . add ( binder )
for model in models : # Now that all models have been given an identifier
# we can write the content to the database .
    try :
        add_pending_model_content ( cursor , publication_id , model )
    except ResourceFileExceededLimitError as e :
        e . publication_id = publication_id
        set_publication_failure ( cursor , e )
return publication_id , insert_mapping


signature:
def _check_pending_document_license_state ( cursor , document_id ) :
    """Check the aggregate state on the pending document ."""

implementation:
cursor . execute ( """\
SELECT BOOL_AND(accepted IS TRUE)
FROM
  pending_documents AS pd,
  license_acceptances AS la
WHERE
  pd.id = %s
  AND
  pd.uuid = la.uuid""" , ( document_id , ) )
try :
    is_accepted = cursor . fetchone ( ) [ 0 ]
except TypeError : # There are no licenses associated with this document .
    is_accepted = True
return is_accepted


signature:
def _check_pending_document_role_state ( cursor , document_id ) :
    """Check the aggregate state on the pending document ."""

implementation:
cursor . execute ( """\
SELECT BOOL_AND(accepted IS TRUE)
FROM
  role_acceptances AS ra,
  pending_documents as pd
WHERE
  pd.id = %s
  AND
  pd.uuid = ra.uuid""" , ( document_id , ) )
try :
    is_accepted = cursor . fetchone ( ) [ 0 ]
except TypeError : # There are no roles to accept
    is_accepted = True
return is_accepted


signature:
def _update_pending_document_state ( cursor , document_id , is_license_accepted , are_roles_accepted ) :
    """Update the state of the document ' s state values ."""

implementation:
args = ( bool ( is_license_accepted ) , bool ( are_roles_accepted ) , document_id , )
cursor . execute ( """\
UPDATE pending_documents
SET (license_accepted, roles_accepted) = (%s, %s)
WHERE id = %s""" , args )


signature:
def is_revision_publication ( publication_id , cursor ) :
    """Checks to see if the publication contains any revised models .
 Revised in this context means that it is a new version of an
 existing piece of content ."""

implementation:
cursor . execute ( """\
SELECT 't'::boolean FROM modules
WHERE uuid IN (SELECT uuid
               FROM pending_documents
               WHERE publication_id = %s)
LIMIT 1""" , ( publication_id , ) )
try :
    cursor . fetchone ( ) [ 0 ]
except TypeError : # NoneType
    has_revision_models = False
else :
    has_revision_models = True
return has_revision_models


signature:
def poke_publication_state ( publication_id , cursor ) :
    """Invoked to poke at the publication to update and acquire its current
 state . This is used to persist the publication to archive ."""

implementation:
cursor . execute ( """\
SELECT "state", "state_messages", "is_pre_publication", "publisher"
FROM publications
WHERE id = %s""" , ( publication_id , ) )
row = cursor . fetchone ( )
current_state , messages , is_pre_publication , publisher = row
if current_state in END_N_INTERIM_STATES : # Bailout early , because the publication is either in progress
# or has been completed .
    return current_state , messages
# Check for acceptance . . .
cursor . execute ( """\
SELECT
  pd.id, license_accepted, roles_accepted
FROM publications AS p JOIN pending_documents AS pd ON p.id = pd.publication_id
WHERE p.id = %s
""" , ( publication_id , ) )
pending_document_states = cursor . fetchall ( )
publication_state_mapping = { }
for document_state in pending_document_states :
    id , is_license_accepted , are_roles_accepted = document_state
    publication_state_mapping [ id ] = [ is_license_accepted , are_roles_accepted ]
    has_changed_state = False
    if is_license_accepted and are_roles_accepted :
        continue
    if not is_license_accepted :
        accepted = _check_pending_document_license_state ( cursor , id )
        if accepted != is_license_accepted :
            has_changed_state = True
            is_license_accepted = accepted
            publication_state_mapping [ id ] [ 0 ] = accepted
    if not are_roles_accepted :
        accepted = _check_pending_document_role_state ( cursor , id )
        if accepted != are_roles_accepted :
            has_changed_state = True
            are_roles_accepted = accepted
            publication_state_mapping [ id ] [ 1 ] = accepted
    if has_changed_state :
        _update_pending_document_state ( cursor , id , is_license_accepted , are_roles_accepted )
# Are all the documents ready for publication ?
state_lump = set ( [ l and r for l , r in publication_state_mapping . values ( ) ] )
is_publish_ready = not ( False in state_lump ) and not ( None in state_lump )
change_state = "Done/Success"
if not is_publish_ready :
    change_state = "Waiting for acceptance"
# Does this publication need moderation ? ( ignore on pre - publication )
# TODO Is this a revision publication ? If so , it doesn ' t matter who the
# user is , because they have been vetted by the previous publisher .
# This has loopholes . . .
if not is_pre_publication and is_publish_ready : # Has this publisher been moderated before ?
    cursor . execute ( """\
SELECT is_moderated
FROM users AS u LEFT JOIN publications AS p ON (u.username = p.publisher)
WHERE p.id = %s""" , ( publication_id , ) )
    try :
        is_publisher_moderated = cursor . fetchone ( ) [ 0 ]
    except TypeError :
        is_publisher_moderated = False
    # Are any of these documents a revision ? Thus vetting of
    # the publisher was done by a vetted peer .
    if not is_publisher_moderated and not is_revision_publication ( publication_id , cursor ) : # Hold up ! This publish needs moderation .
        change_state = "Waiting for moderation"
        is_publish_ready = False
# Publish the pending documents .
if is_publish_ready :
    change_state = "Done/Success"
    if not is_pre_publication :
        publication_state = publish_pending ( cursor , publication_id )
    else :
        cursor . execute ( """\
UPDATE publications
SET state = %s
WHERE id = %s
RETURNING state, state_messages""" , ( change_state , publication_id , ) )
        publication_state , messages = cursor . fetchone ( )
else : # ` change _ state ` set prior to this . . .
    cursor . execute ( """\
UPDATE publications
SET state = %s
WHERE id = %s
RETURNING state, state_messages""" , ( change_state , publication_id , ) )
    publication_state , messages = cursor . fetchone ( )
return publication_state , messages


signature:
def check_publication_state ( publication_id ) :
    """Check the publication ' s current state ."""

implementation:
with db_connect ( ) as db_conn :
    with db_conn . cursor ( ) as cursor :
        cursor . execute ( """\
SELECT "state", "state_messages"
FROM publications
WHERE id = %s""" , ( publication_id , ) )
        publication_state , publication_messages = cursor . fetchone ( )
return publication_state , publication_messages


signature:
def _node_to_model ( tree_or_item , metadata = None , parent = None , lucent_id = cnxepub . TRANSLUCENT_BINDER_ID ) :
    """Given a tree , parse to a set of models"""

implementation:
if 'contents' in tree_or_item : # It is a binder .
    tree = tree_or_item
    binder = cnxepub . TranslucentBinder ( metadata = tree )
    for item in tree [ 'contents' ] :
        node = _node_to_model ( item , parent = binder , lucent_id = lucent_id )
        if node . metadata [ 'title' ] != item [ 'title' ] :
            binder . set_title_for_node ( node , item [ 'title' ] )
    result = binder
else : # It is an item pointing at a document .
    item = tree_or_item
    result = cnxepub . DocumentPointer ( item [ 'id' ] , metadata = item )
if parent is not None :
    parent . append ( result )
return result


signature:
def _reassemble_binder ( id , tree , metadata ) :
    """Reassemble a Binder object coming out of the database ."""

implementation:
binder = cnxepub . Binder ( id , metadata = metadata )
for item in tree [ 'contents' ] :
    node = _node_to_model ( item , parent = binder )
    if node . metadata [ 'title' ] != item [ 'title' ] :
        binder . set_title_for_node ( node , item [ 'title' ] )
return binder


signature:
def publish_pending ( cursor , publication_id ) :
    """Given a publication id as ` ` publication _ id ` ` ,
 write the documents to the * Connexions Archive * ."""

implementation:
cursor . execute ( """\
WITH state_update AS (
  UPDATE publications SET state = 'Publishing' WHERE id = %s
)
SELECT publisher, publication_message
FROM publications
WHERE id = %s""" , ( publication_id , publication_id , ) )
publisher , message = cursor . fetchone ( )
cursor . connection . commit ( )
all_models = [ ]
from . publish import publish_model
# Commit documents one at a time . . .
type_ = cnxepub . Document . __name__
cursor . execute ( """\
SELECT id, uuid, major_version, minor_version, metadata, content
FROM pending_documents
WHERE type = %s AND publication_id = %s""" , ( type_ , publication_id , ) )
for row in cursor . fetchall ( ) : # FIXME Oof , this is hideous !
    id , major_version , minor_version = row [ 1 : 4 ]
    id = str ( id )
    version = '.' . join ( [ str ( x ) for x in ( major_version , minor_version , ) if x is not None ] )
    metadata , content = row [ - 2 : ]
    content = content [ : ]
    metadata [ 'version' ] = version
    document = cnxepub . Document ( id , content , metadata )
    for ref in document . references :
        if ref . uri . startswith ( '/resources/' ) :
            hash = ref . uri [ len ( '/resources/' ) : ]
            cursor . execute ( """\
SELECT data, media_type
FROM pending_resources
WHERE hash = %s""" , ( hash , ) )
            data , media_type = cursor . fetchone ( )
            document . resources . append ( cnxepub . Resource ( hash , io . BytesIO ( data [ : ] ) , media_type , filename = hash ) )
    _ident_hash = publish_model ( cursor , document , publisher , message )
    # noqa
    all_models . append ( document )
# And now the binders , one at a time . . .
type_ = cnxepub . Binder . __name__
cursor . execute ( """\
SELECT id, uuid, major_version, minor_version, metadata, content
FROM pending_documents
WHERE type = %s AND publication_id = %s""" , ( type_ , publication_id , ) )
for row in cursor . fetchall ( ) :
    id , major_version , minor_version , metadata = row [ 1 : 5 ]
    tree = metadata [ '_tree' ]
    binder = _reassemble_binder ( str ( id ) , tree , metadata )
    # Add the resources
    cursor . execute ( """\
SELECT hash, data, media_type, filename
FROM pending_resources r
JOIN pending_resource_associations a ON a.resource_id = r.id
JOIN pending_documents d ON a.document_id = d.id
WHERE ident_hash(uuid, major_version, minor_version) = %s""" , ( binder . ident_hash , ) )
    binder . resources = [ cnxepub . Resource ( r_hash , io . BytesIO ( r_data [ : ] ) , r_media_type , filename = r_filename ) for ( r_hash , r_data , r_media_type , r_filename ) in cursor . fetchall ( ) ]
    _ident_hash = publish_model ( cursor , binder , publisher , message )
    # noqa
    all_models . append ( binder )
# Republish binders containing shared documents .
from . publish import republish_binders
_republished_ident_hashes = republish_binders ( cursor , all_models )
# noqa
# Lastly , update the publication status .
cursor . execute ( """\
UPDATE publications
SET state = 'Done/Success'
WHERE id = %s
RETURNING state""" , ( publication_id , ) )
state = cursor . fetchone ( ) [ 0 ]
return state


signature:
def accept_publication_license ( cursor , publication_id , user_id , document_ids , is_accepted = False ) :
    """Accept or deny the document license for the publication
 ( ` ` publication _ id ` ` ) and user ( at ` ` user _ id ` ` )
 for the documents ( listed by id as ` ` document _ ids ` ` ) ."""

implementation:
cursor . execute ( """\
UPDATE license_acceptances AS la
SET accepted = %s
FROM pending_documents AS pd
WHERE
  pd.publication_id = %s
  AND
  la.user_id = %s
  AND
  pd.uuid = ANY(%s::UUID[])
  AND
  pd.uuid = la.uuid""" , ( is_accepted , publication_id , user_id , document_ids , ) )


signature:
def accept_publication_role ( cursor , publication_id , user_id , document_ids , is_accepted = False ) :
    """Accept or deny the document role attribution for the publication
 ( ` ` publication _ id ` ` ) and user ( at ` ` user _ id ` ` )
 for the documents ( listed by id as ` ` document _ ids ` ` ) ."""

implementation:
cursor . execute ( """\
UPDATE role_acceptances AS ra
SET accepted = %s
FROM pending_documents AS pd
WHERE
  pd.publication_id = %s
  AND
  ra.user_id = %s
  AND
  pd.uuid = ANY(%s::UUID[])
  AND
  pd.uuid = ra.uuid""" , ( is_accepted , publication_id , user_id , document_ids , ) )