<s> from datetime import datetime <EOL> from enum import IntEnum <EOL> from sqlalchemy import Column , Integer , String , Text , ForeignKey , DateTime , or_ , desc <EOL> from sqlalchemy . exc import IntegrityError <EOL> from sqlalchemy . orm import relationship <EOL> from grouper . models . base . model_base import Model <EOL> from grouper . plugin import get_plugins <EOL> class AuditLogCategory ( IntEnum ) : <EOL> """<STR_LIT>""" <EOL> general = <NUM_LIT:1> <EOL> audit = <NUM_LIT:2> <EOL> class AuditLogFailure ( Exception ) : <EOL> pass <EOL> class AuditLog ( Model ) : <EOL> '''<STR_LIT>''' <EOL> __tablename__ = "<STR_LIT>" <EOL> id = Column ( Integer , primary_key = True ) <EOL> log_time = Column ( DateTime , default = datetime . utcnow , nullable = False ) <EOL> actor_id = Column ( Integer , ForeignKey ( "<STR_LIT>" ) , nullable = False ) <EOL> actor = relationship ( "<STR_LIT>" , foreign_keys = [ actor_id ] ) <EOL> on_user_id = Column ( Integer , ForeignKey ( "<STR_LIT>" ) , nullable = True ) <EOL> on_user = relationship ( "<STR_LIT>" , foreign_keys = [ on_user_id ] ) <EOL> on_group_id = Column ( Integer , ForeignKey ( "<STR_LIT>" ) , nullable = True ) <EOL> on_group = relationship ( "<STR_LIT>" , foreign_keys = [ on_group_id ] ) <EOL> on_permission_id = Column ( Integer , ForeignKey ( "<STR_LIT>" ) , nullable = True ) <EOL> on_permission = relationship ( "<STR_LIT>" , foreign_keys = [ on_permission_id ] ) <EOL> action = Column ( String ( length = <NUM_LIT:64> ) , nullable = False ) <EOL> description = Column ( Text , nullable = False ) <EOL> category = Column ( Integer , nullable = False , default = AuditLogCategory . general ) <EOL> @ staticmethod <EOL> def log ( session , actor_id , action , description , <EOL> on_user_id = None , on_group_id = None , on_permission_id = None , <EOL> category = AuditLogCategory . general ) : <EOL> '''<STR_LIT>''' <EOL> entry = AuditLog ( <EOL> actor_id = actor_id , <EOL> log_time = datetime . utcnow ( ) , <EOL> action = action , <EOL> description = description , <EOL> on_user_id = on_user_id if on_user_id else None , <EOL> on_group_id = on_group_id if on_group_id else None , <EOL> on_permission_id = on_permission_id if on_permission_id else None , <EOL> category = int ( category ) , <EOL> ) <EOL> try : <EOL> entry . add ( session ) <EOL> session . flush ( ) <EOL> except IntegrityError : <EOL> session . rollback ( ) <EOL> raise AuditLogFailure ( ) <EOL> session . commit ( ) <EOL> for plugin in get_plugins ( ) : <EOL> plugin . log_auditlog_entry ( entry ) <EOL> @ staticmethod <EOL> def get_entries ( session , actor_id = None , on_user_id = None , on_group_id = None , <EOL> on_permission_id = None , limit = None , offset = None , involve_user_id = None , <EOL> category = None , action = None ) : <EOL> '''<STR_LIT>''' <EOL> results = session . query ( AuditLog ) <EOL> if actor_id : <EOL> results = results . filter ( AuditLog . actor_id == actor_id ) <EOL> if on_user_id : <EOL> results = results . filter ( AuditLog . on_user_id == on_user_id ) <EOL> if on_group_id : <EOL> results = results . filter ( AuditLog . on_group_id == on_group_id ) <EOL> if on_permission_id : <EOL> results = results . filter ( AuditLog . on_permission_id == on_permission_id ) <EOL> if involve_user_id : <EOL> results = results . filter ( or_ ( <EOL> AuditLog . on_user_id == involve_user_id , <EOL> AuditLog . actor_id == involve_user_id <EOL> ) ) <EOL> if category : <EOL> results = results . filter ( AuditLog . category == int ( category ) ) <EOL> if action : <EOL> results = results . filter ( AuditLog . action == action ) <EOL> results = results . order_by ( desc ( AuditLog . log_time ) ) <EOL> if offset : <EOL> results = results . offset ( offset ) <EOL> if limit : <EOL> results = results . limit ( limit ) <EOL> return results . all ( ) </s>
