forked from Netflix/dispatch
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathv1.py
More file actions
133 lines (108 loc) · 3.81 KB
/
v1.py
File metadata and controls
133 lines (108 loc) · 3.81 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
"""
.. module: dispatch.plugins.base.v1
:platform: Unix
:copyright: (c) 2019 by Netflix Inc., see AUTHORS for more
:license: Apache, see LICENSE for more details.
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
"""
import logging
from threading import local
from typing import Any, List, Optional
from pydantic import BaseModel
logger = logging.getLogger(__name__)
class PluginConfiguration(BaseModel):
pass
class IPluginEvent:
name: Optional[str] = None
description: Optional[str] = None
# stolen from https://github.com/getsentry/sentry/
class PluginMount(type):
def __new__(cls, name, bases, attrs):
new_cls = type.__new__(cls, name, bases, attrs)
if IPlugin in bases:
return new_cls
if new_cls.title is None:
new_cls.title = new_cls.__name__
if not new_cls.slug:
new_cls.slug = new_cls.title.replace(" ", "-").lower()
return new_cls
class IPlugin(local):
"""
Plugin interface. Should not be inherited from directly.
A plugin should be treated as if it were a singleton. The owner does not
control when or how the plugin gets instantiated, nor is it guaranteed that
it will happen, or happen more than once.
>>> from dispatch.plugins import Plugin
>>>
>>> class MyPlugin(Plugin):
>>> def get_title(self):
>>> return 'My Plugin'
As a general rule all inherited methods should allow ``**kwargs`` to ensure
ease of future compatibility.
"""
# Generic plugin information
title: Optional[str] = None
slug: Optional[str] = None
description: Optional[str] = None
version: Optional[str] = None
author: Optional[str] = None
author_url: Optional[str] = None
configuration: Optional[dict] = None
project_id: Optional[int] = None
resource_links = ()
schema: PluginConfiguration
commands: List[Any] = []
events: Any = None
plugin_events: Optional[List[IPluginEvent]] = []
# Global enabled state
enabled: bool = False
can_disable: bool = True
multiple: bool = False
def is_enabled(self) -> bool:
"""
Returns a boolean representing if this plugin is enabled.
>>> plugin.is_enabled()
"""
if not self.enabled:
return False
if not self.can_disable:
return True
return True
def get_title(self) -> Optional[str]:
"""
Returns the general title for this plugin.
>>> plugin.get_title()
"""
return self.title
def get_description(self) -> Optional[str]:
"""
Returns the description for this plugin. This is shown on the plugin configuration
page.
>>> plugin.get_description()
"""
return self.description
def get_resource_links(self) -> List[Any]:
"""
Returns a list of tuples pointing to various resources for this plugin.
>>> def get_resource_links(self):
>>> return [
>>> ('Documentation', 'https://dispatch.readthedocs.io'),
>>> ('Bug Tracker', 'https://github.com/Netflix/dispatch/issues'),
>>> ('Source', 'https://github.com/Netflix/dispatch'),
>>> ]
"""
return self.resource_links
def get_event(self, event) -> Optional[IPluginEvent]:
for plugin_event in self.plugin_events:
if plugin_event.slug == event.slug:
return plugin_event
def fetch_incident_events(self, **kwargs):
raise NotImplementedError
class Plugin(IPlugin):
"""
A plugin should be treated as if it were a singleton. The owner does not
control when or how the plugin gets instantiated, nor is it guaranteed that
it will happen, or happen more than once.
"""
__version__ = 1
__metaclass__ = PluginMount