-
Notifications
You must be signed in to change notification settings - Fork 545
Expand file tree
/
Copy pathclient.py
More file actions
73 lines (63 loc) · 2.07 KB
/
client.py
File metadata and controls
73 lines (63 loc) · 2.07 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
from dataclasses import dataclass, field
from plugin.typeform.client import TypeformClient
DEFAULT_CONCURRENCY = 100
DEFAULT_QUEUE_SIZE = 10000
@dataclass
class Spec:
access_token: str
base_url: str = field(default="https://api.typeform.com")
concurrency: int = field(default=DEFAULT_CONCURRENCY)
queue_size: int = field(default=DEFAULT_QUEUE_SIZE)
def validate(self):
if self.access_token is None:
raise Exception("access_token must be provided")
@staticmethod
def json_schema():
return """{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://github.com/cloudquery/cloudquery/plugins/source/typeform/spec",
"$ref": "#/$defs/Spec",
"$defs": {
"Spec": {
"properties": {
"access_token": {
"type": "string",
"minLength": 1,
"description": "Your [personal access token](https://www.typeform.com/developers/get-started/personal-access-token/) from the Typeform Dashboard."
},
"base_url": {
"type": "string",
"default": "https://api.typeform.com",
"description": "The base URL to fetch data from. Use `https://api.eu.typeform.com` if your account is stored in the EU."
},
"concurrency": {
"type": "integer",
"minimum": 1,
"default": 100,
"description": "Maximum number of requests to perform concurrently."
},
"queue_size": {
"type": "integer",
"minimum": 1,
"default": 10000,
"description": "Maximum number of items to have in the queue before waiting for an unfinished request to complete."
}
},
"additionalProperties": false,
"type": "object",
"required": [
"access_token"
]
}
}
}
"""
class Client:
def __init__(self, spec: Spec) -> None:
self._spec = spec
self._client = TypeformClient(spec.access_token, spec.base_url)
def id(self):
return "typeform"
@property
def client(self) -> TypeformClient:
return self._client