File: eventgrid_preparer.py

package info (click to toggle)
python-azure 20250603%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 851,724 kB
  • sloc: python: 7,362,925; ansic: 804; javascript: 287; makefile: 195; sh: 145; xml: 109
file content (46 lines) | stat: -rw-r--r-- 2,663 bytes parent folder | download
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
import functools
from devtools_testutils import EnvironmentVariableLoader

from azure.mgmt.eventgrid.models import Topic, InputSchema, JsonInputSchemaMapping, JsonField, JsonFieldWithDefault


EVENTGRID_TOPIC_PARAM = "eventgrid_topic"
EVENTGRID_TOPIC_LOCATION = "westus"
CLOUD_EVENT_SCHEMA = InputSchema.cloud_event_schema_v1_0
CUSTOM_EVENT_SCHEMA = InputSchema.custom_event_schema
ID_JSON_FIELD = JsonField(source_field="customId")
TOPIC_JSON_FIELD = JsonField(source_field="customTopic")
EVENT_TIME_JSON_FIELD = JsonField(source_field="customEventTime")
EVENT_TYPE_JSON_FIELD_WITH_DEFAULT = JsonFieldWithDefault(source_field="customEventType", default_value="")
SUBJECT_JSON_FIELD_WITH_DEFAULT = JsonFieldWithDefault(source_field="customSubject", default_value="")
DATA_VERSION_JSON_FIELD_WITH_DEFAULT = JsonFieldWithDefault(source_field="customDataVersion", default_value="")
CUSTOM_JSON_INPUT_SCHEMA_MAPPING = JsonInputSchemaMapping(
    id=ID_JSON_FIELD,
    topic=TOPIC_JSON_FIELD,
    event_time=EVENT_TIME_JSON_FIELD,
    event_type=EVENT_TYPE_JSON_FIELD_WITH_DEFAULT,
    subject=SUBJECT_JSON_FIELD_WITH_DEFAULT,
    data_version=DATA_VERSION_JSON_FIELD_WITH_DEFAULT,
)

EventGridPreparer = functools.partial(
    EnvironmentVariableLoader,
    "eventgrid",
    eventgrid_topic_endpoint="https://fakeresource.eastus-1.eventgrid.azure.net/api/events",
    eventgrid_topic_key="fakekeyfakekeyfakekeyfakekeyfakekeyfakekeyA=",
    eventgrid_domain_endpoint="https://fakeresource.eastus-1.eventgrid.azure.net/api/events",
    eventgrid_domain_key="fakekeyfakekeyfakekeyfakekeyfakekeyfakekeyA=",
    eventgrid_cloud_event_topic_endpoint="https://fakeresource.eastus-1.eventgrid.azure.net/api/events",
    eventgrid_cloud_event_topic_key="fakekeyfakekeyfakekeyfakekeyfakekeyfakekeyA=",
    eventgrid_cloud_event_domain_endpoint="https://fakeresource.eastus-1.eventgrid.azure.net/api/events",
    eventgrid_cloud_event_domain_key="fakekeyfakekeyfakekeyfakekeyfakekeyfakekeyA=",
    eventgrid_custom_event_topic_endpoint="https://fakeresource.eastus-1.eventgrid.azure.net/api/events",
    eventgrid_custom_event_topic_key="fakekeyfakekeyfakekeyfakekeyfakekeyfakekeyA=",
    eventgrid_partner_channel_name="fakechannel",
    eventgrid_partner_namespace_topic_endpoint="https://fakeresource.eastus-1.eventgrid.azure.net/api/events",
    eventgrid_partner_namespace_topic_key="fakekeyfakekeyfakekeyfakekeyfakekeyfakekeyA=",
    eventgrid_endpoint="https://fakeresource.eastus-1.eventgrid.azure.net/api/events",
    eventgrid_key="fakekeyfakekeyfakekeyfakekeyfakekeyfakekeyA=",
    eventgrid_topic_name="faketopic",
    eventgrid_event_subscription_name="fakesub",
)