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
|
import requests
import json
##############################
def extract_snippet_fields(flow):
return {
key: flow.get(key, [])
for key in [
"processors",
"connections",
"funnels",
"inputPorts",
"outputPorts",
"labels",
"controllerServices",
"processGroups"
]
}
def upload_nifi_exported_flow(
nifi_host: str,
username: str,
password: str,
json_file_path: str,
verify_ssl: bool = False
):
try:
token_resp = requests.post(
f"{nifi_host}/nifi-api/access/token",
data={"username": username, "password": password},
verify=verify_ssl
)
token_resp.raise_for_status()
token = token_resp.text
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
root_resp = requests.get(f"{nifi_host}/nifi-api/flow/process-groups/root", headers=headers, verify=verify_ssl)
root_resp.raise_for_status()
root_pg_id = root_resp.json()["processGroupFlow"]["id"]
with open(json_file_path, "r") as f:
raw = json.load(f)
flow = raw.get("flowContents")
if not flow:
raise ValueError("Missing 'flowContents' in provided file.")
snippet = extract_snippet_fields(flow)
payload = {
"revision": { "version": 0 },
"component": {
"name": flow.get("name", "ImportedGroup"),
"position": { "x": 0.0, "y": 0.0 },
"flowSnippet": snippet
}
}
url = f"{nifi_host}/nifi-api/process-groups/{root_pg_id}/process-groups"
resp = requests.post(url, headers=headers, json=payload, verify=verify_ssl)
if resp.status_code == 201:
print("✅ Flow uploaded successfully!")
pg_id = resp.json()["component"]["id"]
print(f"🔗 View it at: {nifi_host}/nifi/#/process-groups/{root_pg_id}/process-group/{pg_id}")
else:
print(f"❌ Upload failed: {resp.status_code} - {resp.text}")
except Exception as e:
print(f"🚨 Error: {e}")
|
