Set Up a Hot Folder
Watch a directory so new files transfer automatically.
A hot folder watches a directory on a device; files dropped there are picked up and transferred automatically. This example registers a hot folder and lists existing ones.
Steps
- Log in and choose the device and directory to watch.
- Register the hot folder with
POST /api/hot-folder. - List hot folders with
GET /api/hot-folder. - Update or delete a hot folder by its ID as needed.
Configuration
Set your credentials and IDs via environment variables or a .env file (INNORIX_BASE_URL, INNORIX_WORKSPACE_ID, INNORIX_EMAIL, INNORIX_PASSWORD, and the INNORIX_SOURCE_ID / INNORIX_TARGET_ID / path variables the example uses).
API
| Method | Endpoint | Description |
|---|---|---|
| POST | /api/hot-folder | Create a hot folder |
| GET | /api/hot-folder | List hot folders |
| GET | /api/hot-folder/{id} | Get a hot folder |
| PATCH | /api/hot-folder/{id} | Update a hot folder |
| DELETE | /api/hot-folder/{id} | Delete a hot folder |
python
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
134
"""
08 - Hot folders
===============
A hot folder watches a directory on a device so that files placed there are
picked up automatically. This example registers a hot folder, lists them, and
shows how to update/delete one.
This file is self-contained: it embeds a minimal client so it can be read and
run on its own.
Endpoints:
POST /api/hot-folder
GET /api/hot-folder
GET /api/hot-folder/{id}
PATCH /api/hot-folder/{id}
DELETE /api/hot-folder/{id}
Run:
python 08_hot_folder.py
"""
from __future__ import annotations
import os
import json
import logging
import requests
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
pass
# --------------------------------------------------------------------------- #
# Configuration
# --------------------------------------------------------------------------- #
BASE_URL = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com")
WORKSPACE_ID = os.getenv("INNORIX_WORKSPACE_ID", "<WORKSPACE_ID>")
EMAIL = os.getenv("INNORIX_EMAIL", "<YOUR_EMAIL>")
PASSWORD = os.getenv("INNORIX_PASSWORD", "<YOUR_PASSWORD>")
SOURCE_ID = os.getenv("INNORIX_SOURCE_ID", "<SOURCE_DEVICE_ID>")
SOURCE_PATHS = [p.strip() for p in os.getenv(
"INNORIX_SOURCE_PATHS", "C:/Users/innorix/Downloads/image").split(",") if p.strip()]
logging.basicConfig(level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%H:%M:%S")
log = logging.getLogger("innorix")
# --------------------------------------------------------------------------- #
# Minimal client
# --------------------------------------------------------------------------- #
class ApiError(RuntimeError):
"""Raised when the API returns a non-2xx response."""
class InnorixClient:
def __init__(self, base_url: str, workspace_id: str, timeout: float = 15.0):
self.base_url = base_url.rstrip("/")
self.workspace_id = workspace_id
self.timeout = timeout
self.session = requests.Session()
def _request(self, method: str, path: str, **kwargs):
kwargs.setdefault("timeout", self.timeout)
res = self.session.request(method, f"{self.base_url}{path}", **kwargs)
try:
res.raise_for_status()
except requests.HTTPError as exc:
raise ApiError(f"{method} {path} -> {res.status_code}: {res.text[:500]}") from exc
if not res.content:
return None
try:
return res.json()
except ValueError:
return res.text
def login(self, email: str, password: str) -> str:
data = self._request("POST", "/api/auth/login",
json={"email": email, "password": password}, timeout=10)
token = data["data"]["user"]["accessToken"]
self.session.headers.update({
"Authorization": f"Bearer {token}",
"x-workspace-id": self.workspace_id,
"Content-Type": "application/json",
})
log.info("logged in")
return token
def create_hot_folder(self, device_id: str, path: str):
return self._request("POST", "/api/hot-folder",
json={"deviceId": device_id, "path": path})
def list_hot_folders(self):
return self._request("GET", "/api/hot-folder", timeout=10)
def get_hot_folder(self, hot_folder_id: str):
return self._request("GET", f"/api/hot-folder/{hot_folder_id}", timeout=10)
def update_hot_folder(self, hot_folder_id: str, patch: dict):
return self._request("PATCH", f"/api/hot-folder/{hot_folder_id}", json=patch)
def delete_hot_folder(self, hot_folder_id: str):
return self._request("DELETE", f"/api/hot-folder/{hot_folder_id}", timeout=10)
# --------------------------------------------------------------------------- #
# Run
# --------------------------------------------------------------------------- #
def main() -> None:
client = InnorixClient(BASE_URL, WORKSPACE_ID)
client.login(EMAIL, PASSWORD)
# Create a hot folder on the source device.
created = client.create_hot_folder(device_id=SOURCE_ID, path=SOURCE_PATHS[0])
print(json.dumps(created, indent=2, ensure_ascii=False))
# List existing hot folders.
folders = client.list_hot_folders()
print(json.dumps(folders, indent=2, ensure_ascii=False)[:1000])
# Update / delete a specific hot folder by its UUID:
# hot_folder_id = "<HOT_FOLDER_ID>"
# client.get_hot_folder(hot_folder_id)
# client.update_hot_folder(hot_folder_id, {"path": "C:/data/new_watch"})
# client.delete_hot_folder(hot_folder_id)
if __name__ == "__main__":
main()