event-tracking/tracker.py at master - GitHub

Track application events. Supports persisting events to multiple backends. Best Practices: * It is recommended that event types are namespaced using dot ... Skiptocontent {{message}} openedx / event-tracking Public Notifications Fork 34 Star 51 Code Issues 0 Pullrequests 1 Actions Projects 0 Wiki Security Insights More Code Issues Pullrequests Actions Projects Wiki Security Insights Permalink master Branches Tags Couldnotloadbranches Nothingtoshow {{refName}} default Couldnotloadtags Nothingtoshow {{refName}} default event-tracking/eventtracking/tracker.py / Jumpto Tracker Class __init__ Function located_context Function get_backend Function processors Function backends Function emit Function resolve_context Function enter_context Function exit_context Function context Function register_tracker Function get_tracker Function emit Function Gotofile Gotofile T Gotoline L Gotodefinition R Copypath Copypermalink Thiscommitdoesnotbelongtoanybranchonthisrepository,andmaybelongtoaforkoutsideoftherepository.     Cannotretrievecontributorsatthistime 143lines(116sloc) 4.55KB Raw Blame Editthisfile E OpeninGitHubDesktop OpenwithDesktop Viewraw Viewblame ThisfilecontainsbidirectionalUnicodetextthatmaybeinterpretedorcompileddifferentlythanwhatappearsbelow.Toreview,openthefileinaneditorthatrevealshiddenUnicodecharacters. LearnmoreaboutbidirectionalUnicodecharacters Showhiddencharacters """ Trackapplicationevents.Supportspersistingeventstomultiplebackends. BestPractices: *Itisrecommendedthateventtypesarenamespacedusingdotnotationto avoidnamingcollisions,similartoDNSnames.Forexample: org.edx.video.stop,edu.mit.audio.stop *Avoidusingeventtypenamesthatmaycausecollisions.Theburdenis ontheanalysttodecidewhetheryoureventisequivalenttoanother andshouldbegroupedaccordinglyetc. *Donotemiteventsthatyoudon'town.Thiscouldnegativelyimpact theanalysisoftheeventstream.Ifyoususpectyoureventis equivalenttoanother,saysoinyourdocumenation,andtheanalyst candecidewhetherornottogroupthem. """ fromcontextlibimportcontextmanager fromdatetimeimportdatetime importlogging frompytzimportUTC fromeventtracking.locatorimportDefaultContextLocator fromeventtracking.backends.routingimportRoutingBackend UNKNOWN_EVENT_TYPE='unknown' DEFAULT_TRACKER_NAME='default' TRACKERS={} LOG=logging.getLogger(__name__) classTracker: """ Trackapplicationevents.Holdsreferencestoasetofbackendsthatwill beusedtopersistanyeventsthatareemitted. """ def__init__(self,backends=None,context_locator=None,processors=None): self.routing_backend=RoutingBackend(backends=backends,processors=processors) self.context_locator=context_locatororDefaultContextLocator() @property deflocated_context(self): """ Thethreadlocalcontextforthistracker. """ returnself.context_locator.get() defget_backend(self,name): """Getsthebackendthatwasconfiguredwith`name`""" returnself.backends[name] @property defprocessors(self): """Thelistofregisteredprocessors""" returnself.routing_backend.processors @property defbackends(self): """Thedictionaryofregisteredbackends""" returnself.routing_backend.backends defemit(self,name=None,data=None): """ EmitaneventannotatedwiththeUTCtimewhenthisfunctionwascalled. `name`isauniqueidentificationstringforaneventthathas alreadybeenregistered. `data`isadictionarymappingfieldnamestothevaluetoincludeintheevent. Notethatallvaluesprovidedmustbeserializable. """ event={ 'name':nameorUNKNOWN_EVENT_TYPE, 'timestamp':datetime.now(UTC), 'data':dataor{}, 'context':self.resolve_context() } self.routing_backend.send(event) defresolve_context(self): """ Createanewdictionarythatcorrespondstotheunionofallofthe contextsthathavebeenenteredbutnotexitedatthispoint. """ merged={} forcontextinself.located_context.values(): merged.update(context) returnmerged defenter_context(self,name,ctx): """ Enteranamedcontext.Anyeventsemittedaftercallingthis methodwillcontainallofthekey-valuepairsincludedin`ctx` unlessoverriddenbyacontextthatisenteredafterthiscall. """ self.located_context[name]=ctx defexit_context(self,name): """ Exitanamedcontext.Thiswillremoveallkey-valuepairs associatedwiththiscontextfromanyeventsemittedafterit isremoved. """ delself.located_context[name] @contextmanager defcontext(self,name,ctx): """ Executetheblockwiththegivencontextapplied.Thismanager ensuresthatthecontextisremovedevenifanexceptionisraised withinthecontext. """ self.enter_context(name,ctx) try: yield finally: self.exit_context(name) defregister_tracker(tracker,name=DEFAULT_TRACKER_NAME): """ Makesatrackergloballyaccessible.Providingno`name`parameter allowsyoutoregistertheglobaldefaulttrackerthatwillbeused bysubsequentcallsto`tracker.emit`. """ TRACKERS[name]=tracker defget_tracker(name=DEFAULT_TRACKER_NAME): """ Getsanamedtracker.Defaultstothedefaultglobaltracker.Raises a`KeyError`ifnosuchtrackerhasbeenregisteredbypreviouslycalling `register_tracker`. """ returnTRACKERS[name] defemit(name=None,data=None): """Calls`Tracker.emit`onthedefaultglobaltracker""" returnget_tracker().emit(name=name,data=data) Copylines Copypermalink Viewgitblame Referenceinnewissue Go Youcan’tperformthatactionatthistime. Yousignedinwithanothertaborwindow.Reloadtorefreshyoursession. Yousignedoutinanothertaborwindow.Reloadtorefreshyoursession.