FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
aws-lambda-builders/aws_lambda_builders/registry.py at develop · QPC-database/aws-lambda-builders · GitHub
QPC-database
/
aws-lambda-builders
Public
forked from
aws/aws-lambda-builders
Notifications
You must be signed in to change notification settings
Fork
0
Star
1
Code
Pull requests
0
Actions
Projects
Security and quality
0
Insights
Additional navigation options
Code
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
aws-lambda-builders
/
aws_lambda_builders
/
registry.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
99 lines (71 loc) · 3.07 KB
Breadcrumbs
aws-lambda-builders
/
aws_lambda_builders
/
registry.py
Copy path
File metadata and controls
99 lines (71 loc) · 3.07 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
"""
Registry of auto-discovered workflows. Classes and methods in this module are thread-safe.
"""
import
threading
from
aws_lambda_builders
.
exceptions
import
WorkflowNotFoundError
class
Registry
(
object
):
"""
Stores a registry of workflows. Only one workflow matching a given capability can be stored. If attempting to
register two workflows for same capability, it will raise an exception.
This class is thread-safe for multiple writers (ie. registers happening simultaneously from different threads).
"""
def
__init__
(
self
,
write_lock
=
None
):
self
.
_write_lock
=
write_lock
or
threading
.
Lock
()
self
.
_data
=
{}
def
__getitem__
(
self
,
capability
):
key
=
self
.
_make_key
(
capability
)
return
self
.
_data
[
key
]
def
__setitem__
(
self
,
capability
,
value
):
key
=
self
.
_make_key
(
capability
)
try
:
self
.
_write_lock
.
acquire
()
if
key
in
self
.
_data
:
raise
KeyError
(
"A workflow with given capabilities '{}' is already registered"
.
format
(
key
))
self
.
_data
[
key
]
=
value
finally
:
self
.
_write_lock
.
release
()
def
__contains__
(
self
,
capability
):
key
=
self
.
_make_key
(
capability
)
return
key
in
self
.
_data
def
__len__
(
self
):
return
len
(
self
.
_data
)
def
clear
(
self
):
try
:
self
.
_write_lock
.
acquire
()
self
.
_data
.
clear
()
finally
:
self
.
_write_lock
.
release
()
@
staticmethod
def
_make_key
(
capability
):
"""
Given the capabilities, generate a string that can be used as the key for the registry object
"""
# Key is created by concatenating the capabilites data with underscore.
# This delimiter is positional ie. if a value is not provided, the delimiter still needs to exist in the key.
# This helps us be forwards compatible with new capabilities
return
"_"
.
join
(
[
capability
.
language
or
""
,
capability
.
dependency_manager
or
""
,
capability
.
application_framework
or
""
]
).
lower
()
# Built-in registry of workflows.
DEFAULT_REGISTRY
=
Registry
()
def
get_workflow
(
capability
,
registry
=
DEFAULT_REGISTRY
):
"""
Find and return a workflow class capable of acting on the given combination of capabilities
:type capability: aws_lambda_builders.workflow.capabilities
:param capability:
The capabilities you want from this workflow
:type registry: aws_lambda_builders.registry.Registry
:param registry:
Registry to fetch the workflow from
:rtype: aws_lambda_builders.workflow.BaseWorkflow
:return:
A workflow class that has the given capabilities. Note: This returns a class and not an instance of the class.
:raises WorkflowNotFoundError: If a workflow with given capabilities was not found
"""
if
capability
not
in
registry
:
raise
WorkflowNotFoundError
(
language
=
capability
.
language
,
dependency_manager
=
capability
.
dependency_manager
,
application_framework
=
capability
.
application_framework
,
)
return
registry
[
capability
]
Back
|
FazBrowse Home
|
New Git URL