API Reference

strapi_client

StrapiClientAsync

Bases: StrapiClientBase

Async REST API client for Strapi.

Source code in src/strapi_client/strapi_client_async.py 
 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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
class StrapiClientAsync(StrapiClientBase):
    """Async REST API client for Strapi."""

    _client: httpx.AsyncClient | None = None

    async def __aenter__(self):
        self.open()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.close()

    def open(self) -> httpx.AsyncClient:
        # Fallback to creating a client if not used in a context manager.
        if self._client is None:
            self._client = httpx.AsyncClient(timeout=self.timeout)
        return self._client

    async def close(self):
        if self._client is not None:
            await self._client.aclose()
            self._client = None

    @property
    def client(self) -> httpx.AsyncClient:
        if not self._client:
            raise RuntimeError("Client is not initialized.")
        return self._client

    async def authorize(
        self,
        identifier: str,
        password: str,
    ) -> None:
        """Get auth token using identifier and password."""
        res = await self.send_post_request(
            "auth/local", json=AuthPayload(identifier=identifier, password=password).model_dump(), use_auth=False
        )
        self._token = AuthResponse.model_validate(res.json()).jwt

    async def get_single_document(self, single_api_id: str) -> DocumentResponse:
        """Get document of single type."""
        res = await self.send_get_request(single_api_id)
        return DocumentResponse.model_validate(res.json())

    async def get_document(
        self,
        plural_api_id: str,
        document_id: str,
        populate: list[str] | dict[str, Any] | str | None = None,
        fields: list[str] | None = None,
        locale: str | None = None,
    ) -> DocumentResponse:
        """Get document by document id."""
        params = ApiParameters(populate=populate, fields=fields, locale=locale)
        res = await self.send_get_request(f"{plural_api_id}/{document_id}", params=params.stringify())
        return DocumentResponse.model_validate(res.json())

    async def get_documents(
        self,
        plural_api_id: str,
        sort: list[str] | None = None,
        filters: dict[str, Any] | None = None,
        populate: list[str] | dict[str, Any] | str | None = None,
        fields: list[str] | None = None,
        publication_state: str | None = None,
        locale: str | None = None,
        start: int | None = None,
        page: int | None = None,
        batch_size: int = 25,
        with_count: bool = True,
    ) -> DocumentsResponse:
        """Get list of documents. By default, operates in batch mode to get all documents automatically."""
        params = ApiParameters(
            sort=sort,
            filters=filters,
            populate=populate,
            fields=fields,
            page=page,
            page_size=batch_size,
            start=start,
            limit=batch_size,
            publication_state=publication_state,
            locale=locale,
        )
        if params.page is not None or params.start is not None:  # Get specific page/batch
            res = await self.send_get_request(plural_api_id, params=params.stringify())
            return DocumentsResponse.model_validate(res.json())
        else:  # Get all records
            params.start = 0
            params.with_count = True
            res = await self.send_get_request(plural_api_id, params=params.stringify())
            res_page = DocumentsResponse.model_validate(res.json())
            start_list = [i for i in range(batch_size, res_page.meta.get_total_count(), batch_size)]
            all_data = res_page
            for cur_start in start_list:
                params.start = cur_start
                params.with_count = with_count
                res = await self.send_get_request(plural_api_id, params=params.stringify())
                res_page = DocumentsResponse.model_validate(res.json())
                all_data.data += res_page.data
                all_data.meta = res_page.meta
            return all_data

    async def create_or_update_single_document(
        self, single_api_id: str, data: dict[str, Any] | BaseModel
    ) -> DocumentResponse:
        """Create or update single type document."""
        res = await self.send_put_request(single_api_id, body={"data": serialize_document_data(data)})
        return DocumentResponse.model_validate(res.json())

    async def create_document(self, plural_api_id: str, data: dict[str, Any] | BaseModel) -> DocumentResponse:
        """Create new document."""
        res = await self.send_post_request(
            plural_api_id,
            json={"data": serialize_document_data(data)},
        )
        return DocumentResponse.model_validate(res.json())

    async def update_document(
        self, plural_api_id: str, document_id: str, data: dict[str, Any] | BaseModel
    ) -> DocumentResponse:
        """Update document fields."""
        res = await self.send_put_request(
            f"{plural_api_id}/{document_id}",
            body={"data": serialize_document_data(data)},
        )
        return DocumentResponse.model_validate(res.json())

    async def delete_single_document(self, single_api_id: str) -> None:
        """Delete single type document."""
        await self.send_delete_request(single_api_id)

    async def delete_document(self, plural_api_id: str, document_id: str) -> None:
        """Delete document by document id."""
        await self.send_delete_request(f"{plural_api_id}/{document_id}")

    async def send_get_request(
        self,
        route: str,
        params: dict[str, Any] | str | None = None,
        use_auth: bool = True,
    ) -> httpx.Response:
        """Send GET request to custom endpoint."""
        res = await self.client.get(
            url=urljoin(self.api_url, route), params=params, headers=self._auth_header if use_auth else None
        )
        self._check_response(res, "Unable to send GET request")
        return res

    async def send_put_request(
        self,
        route: str,
        body: dict[str, Any] | None = None,
        params: dict[str, Any] | str | None = None,
        use_auth: bool = True,
    ) -> httpx.Response:
        """Send PUT request to custom endpoint."""
        res = await self.client.put(
            url=urljoin(self.api_url, route), json=body, params=params, headers=self._auth_header if use_auth else None
        )
        self._check_response(res, "Unable to send PUT request")
        return res

    async def send_post_request(
        self,
        route: str,
        json: dict[str, Any] | None = None,
        params: dict[str, Any] | str | None = None,
        data: dict[str, Any] | None = None,
        files: list | None = None,
        use_auth: bool = True,
    ) -> httpx.Response:
        """Send POST request to custom endpoint."""
        res = await self.client.post(
            url=urljoin(self.api_url, route),
            json=json,
            params=params,
            data=data,
            files=files,
            headers=self._auth_header if use_auth else None,
        )
        self._check_response(res, "Unable to send POST request")
        return res

    async def send_delete_request(self, route: str, use_auth: bool = True) -> httpx.Response:
        """Send DELETE request to custom endpoint."""
        res = await self.client.delete(
            url=urljoin(self.api_url, route), headers=self._auth_header if use_auth else None
        )
        self._check_response(res, "Unable to send DELETE request")
        return res

    async def upload_files(
        self,
        files: list[Path | str] | dict[str, BytesIO | bytes | bytearray | memoryview],
        content_type_id: str | None = None,
        document_id: int | str | None = None,
        field: str | None = None,
    ) -> list[MediaImageDocument]:
        """Upload a list of files."""
        file_payloads = FilePayload.list_from_files(files)
        data: dict[str, Any] = {}
        if content_type_id and document_id and field:
            data = {"ref": content_type_id, "refId": document_id, "field": field}
        res = await self.send_post_request("upload", data=data, files=[fp.to_files_tuple() for fp in file_payloads])
        self._check_response(res, "Unable to send POST request")
        return [MediaImageDocument.model_validate(d) for d in (res.json() or [])]

    async def upload_file(
        self,
        file: Path | str | dict[str, BytesIO | bytes | bytearray | memoryview],
        content_type_id: str | None = None,
        document_id: int | str | None = None,
        field: str | None = None,
    ) -> MediaImageDocument:
        """Upload a list of files."""
        if isinstance(file, dict):
            if len(file) > 1:
                raise ValueError("One file must be provided in binary dict")
            files: list[Path | str] | dict[str, BytesIO | bytes | bytearray | memoryview] = file
        else:
            files = [file]
        result = await self.upload_files(
            files=files,
            content_type_id=content_type_id,
            document_id=document_id,
            field=field,
        )
        if not result or len(result) != 1:
            raise ValueError("One and only one result is expected after operation")
        return result[0]

    async def get_uploaded_files(self, filters: dict | None = None) -> list[dict[str, Any]]:
        """Get uploaded files."""
        params = ApiParameters(filters=filters)
        res = await self.send_get_request("upload/files", params=params.stringify())
        return res.json()

    async def check_health(self) -> bool:
        """Check if Strapi API is available."""
        try:
            async with httpx.AsyncClient(timeout=5.0) as client:
                res = await client.get(urljoin(self.base_url, "_health"))
                res.raise_for_status()
                return True
        except Exception:
            return False

authorize async 

authorize(identifier, password)

Get auth token using identifier and password.

Source code in src/strapi_client/strapi_client_async.py 
45
46
47
48
49
50
51
52
53
54
async def authorize(
    self,
    identifier: str,
    password: str,
) -> None:
    """Get auth token using identifier and password."""
    res = await self.send_post_request(
        "auth/local", json=AuthPayload(identifier=identifier, password=password).model_dump(), use_auth=False
    )
    self._token = AuthResponse.model_validate(res.json()).jwt

get_single_document async 

get_single_document(single_api_id)

Get document of single type.

Source code in src/strapi_client/strapi_client_async.py 
56
57
58
59
async def get_single_document(self, single_api_id: str) -> DocumentResponse:
    """Get document of single type."""
    res = await self.send_get_request(single_api_id)
    return DocumentResponse.model_validate(res.json())

get_document async 

get_document(
    plural_api_id,
    document_id,
    populate=None,
    fields=None,
    locale=None,
)

Get document by document id.

Source code in src/strapi_client/strapi_client_async.py 
61
62
63
64
65
66
67
68
69
70
71
72
async def get_document(
    self,
    plural_api_id: str,
    document_id: str,
    populate: list[str] | dict[str, Any] | str | None = None,
    fields: list[str] | None = None,
    locale: str | None = None,
) -> DocumentResponse:
    """Get document by document id."""
    params = ApiParameters(populate=populate, fields=fields, locale=locale)
    res = await self.send_get_request(f"{plural_api_id}/{document_id}", params=params.stringify())
    return DocumentResponse.model_validate(res.json())

get_documents async 

get_documents(
    plural_api_id,
    sort=None,
    filters=None,
    populate=None,
    fields=None,
    publication_state=None,
    locale=None,
    start=None,
    page=None,
    batch_size=25,
    with_count=True,
)

Get list of documents. By default, operates in batch mode to get all documents automatically.

Source code in src/strapi_client/strapi_client_async.py 
 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
async def get_documents(
    self,
    plural_api_id: str,
    sort: list[str] | None = None,
    filters: dict[str, Any] | None = None,
    populate: list[str] | dict[str, Any] | str | None = None,
    fields: list[str] | None = None,
    publication_state: str | None = None,
    locale: str | None = None,
    start: int | None = None,
    page: int | None = None,
    batch_size: int = 25,
    with_count: bool = True,
) -> DocumentsResponse:
    """Get list of documents. By default, operates in batch mode to get all documents automatically."""
    params = ApiParameters(
        sort=sort,
        filters=filters,
        populate=populate,
        fields=fields,
        page=page,
        page_size=batch_size,
        start=start,
        limit=batch_size,
        publication_state=publication_state,
        locale=locale,
    )
    if params.page is not None or params.start is not None:  # Get specific page/batch
        res = await self.send_get_request(plural_api_id, params=params.stringify())
        return DocumentsResponse.model_validate(res.json())
    else:  # Get all records
        params.start = 0
        params.with_count = True
        res = await self.send_get_request(plural_api_id, params=params.stringify())
        res_page = DocumentsResponse.model_validate(res.json())
        start_list = [i for i in range(batch_size, res_page.meta.get_total_count(), batch_size)]
        all_data = res_page
        for cur_start in start_list:
            params.start = cur_start
            params.with_count = with_count
            res = await self.send_get_request(plural_api_id, params=params.stringify())
            res_page = DocumentsResponse.model_validate(res.json())
            all_data.data += res_page.data
            all_data.meta = res_page.meta
        return all_data

create_or_update_single_document async 

create_or_update_single_document(single_api_id, data)

Create or update single type document.

Source code in src/strapi_client/strapi_client_async.py 
120
121
122
123
124
125
async def create_or_update_single_document(
    self, single_api_id: str, data: dict[str, Any] | BaseModel
) -> DocumentResponse:
    """Create or update single type document."""
    res = await self.send_put_request(single_api_id, body={"data": serialize_document_data(data)})
    return DocumentResponse.model_validate(res.json())

create_document async 

create_document(plural_api_id, data)

Create new document.

Source code in src/strapi_client/strapi_client_async.py 
127
128
129
130
131
132
133
async def create_document(self, plural_api_id: str, data: dict[str, Any] | BaseModel) -> DocumentResponse:
    """Create new document."""
    res = await self.send_post_request(
        plural_api_id,
        json={"data": serialize_document_data(data)},
    )
    return DocumentResponse.model_validate(res.json())

update_document async 

update_document(plural_api_id, document_id, data)

Update document fields.

Source code in src/strapi_client/strapi_client_async.py 
135
136
137
138
139
140
141
142
143
async def update_document(
    self, plural_api_id: str, document_id: str, data: dict[str, Any] | BaseModel
) -> DocumentResponse:
    """Update document fields."""
    res = await self.send_put_request(
        f"{plural_api_id}/{document_id}",
        body={"data": serialize_document_data(data)},
    )
    return DocumentResponse.model_validate(res.json())

delete_single_document async 

delete_single_document(single_api_id)

Delete single type document.

Source code in src/strapi_client/strapi_client_async.py 
145
146
147
async def delete_single_document(self, single_api_id: str) -> None:
    """Delete single type document."""
    await self.send_delete_request(single_api_id)

delete_document async 

delete_document(plural_api_id, document_id)

Delete document by document id.

Source code in src/strapi_client/strapi_client_async.py 
149
150
151
async def delete_document(self, plural_api_id: str, document_id: str) -> None:
    """Delete document by document id."""
    await self.send_delete_request(f"{plural_api_id}/{document_id}")

send_get_request async 

send_get_request(route, params=None, use_auth=True)

Send GET request to custom endpoint.

Source code in src/strapi_client/strapi_client_async.py 
153
154
155
156
157
158
159
160
161
162
163
164
async def send_get_request(
    self,
    route: str,
    params: dict[str, Any] | str | None = None,
    use_auth: bool = True,
) -> httpx.Response:
    """Send GET request to custom endpoint."""
    res = await self.client.get(
        url=urljoin(self.api_url, route), params=params, headers=self._auth_header if use_auth else None
    )
    self._check_response(res, "Unable to send GET request")
    return res

send_put_request async 

send_put_request(
    route, body=None, params=None, use_auth=True
)

Send PUT request to custom endpoint.

Source code in src/strapi_client/strapi_client_async.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
async def send_put_request(
    self,
    route: str,
    body: dict[str, Any] | None = None,
    params: dict[str, Any] | str | None = None,
    use_auth: bool = True,
) -> httpx.Response:
    """Send PUT request to custom endpoint."""
    res = await self.client.put(
        url=urljoin(self.api_url, route), json=body, params=params, headers=self._auth_header if use_auth else None
    )
    self._check_response(res, "Unable to send PUT request")
    return res

send_post_request async 

send_post_request(
    route,
    json=None,
    params=None,
    data=None,
    files=None,
    use_auth=True,
)

Send POST request to custom endpoint.

Source code in src/strapi_client/strapi_client_async.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
async def send_post_request(
    self,
    route: str,
    json: dict[str, Any] | None = None,
    params: dict[str, Any] | str | None = None,
    data: dict[str, Any] | None = None,
    files: list | None = None,
    use_auth: bool = True,
) -> httpx.Response:
    """Send POST request to custom endpoint."""
    res = await self.client.post(
        url=urljoin(self.api_url, route),
        json=json,
        params=params,
        data=data,
        files=files,
        headers=self._auth_header if use_auth else None,
    )
    self._check_response(res, "Unable to send POST request")
    return res

send_delete_request async 

send_delete_request(route, use_auth=True)

Send DELETE request to custom endpoint.

Source code in src/strapi_client/strapi_client_async.py 
201
202
203
204
205
206
207
async def send_delete_request(self, route: str, use_auth: bool = True) -> httpx.Response:
    """Send DELETE request to custom endpoint."""
    res = await self.client.delete(
        url=urljoin(self.api_url, route), headers=self._auth_header if use_auth else None
    )
    self._check_response(res, "Unable to send DELETE request")
    return res

upload_files async 

upload_files(
    files,
    content_type_id=None,
    document_id=None,
    field=None,
)

Upload a list of files.

Source code in src/strapi_client/strapi_client_async.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
async def upload_files(
    self,
    files: list[Path | str] | dict[str, BytesIO | bytes | bytearray | memoryview],
    content_type_id: str | None = None,
    document_id: int | str | None = None,
    field: str | None = None,
) -> list[MediaImageDocument]:
    """Upload a list of files."""
    file_payloads = FilePayload.list_from_files(files)
    data: dict[str, Any] = {}
    if content_type_id and document_id and field:
        data = {"ref": content_type_id, "refId": document_id, "field": field}
    res = await self.send_post_request("upload", data=data, files=[fp.to_files_tuple() for fp in file_payloads])
    self._check_response(res, "Unable to send POST request")
    return [MediaImageDocument.model_validate(d) for d in (res.json() or [])]

upload_file async 

upload_file(
    file, content_type_id=None, document_id=None, field=None
)

Upload a list of files.

Source code in src/strapi_client/strapi_client_async.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
async def upload_file(
    self,
    file: Path | str | dict[str, BytesIO | bytes | bytearray | memoryview],
    content_type_id: str | None = None,
    document_id: int | str | None = None,
    field: str | None = None,
) -> MediaImageDocument:
    """Upload a list of files."""
    if isinstance(file, dict):
        if len(file) > 1:
            raise ValueError("One file must be provided in binary dict")
        files: list[Path | str] | dict[str, BytesIO | bytes | bytearray | memoryview] = file
    else:
        files = [file]
    result = await self.upload_files(
        files=files,
        content_type_id=content_type_id,
        document_id=document_id,
        field=field,
    )
    if not result or len(result) != 1:
        raise ValueError("One and only one result is expected after operation")
    return result[0]

get_uploaded_files async 

get_uploaded_files(filters=None)

Get uploaded files.

Source code in src/strapi_client/strapi_client_async.py 
249
250
251
252
253
async def get_uploaded_files(self, filters: dict | None = None) -> list[dict[str, Any]]:
    """Get uploaded files."""
    params = ApiParameters(filters=filters)
    res = await self.send_get_request("upload/files", params=params.stringify())
    return res.json()

check_health async 

check_health()

Check if Strapi API is available.

Source code in src/strapi_client/strapi_client_async.py 
255
256
257
258
259
260
261
262
263
async def check_health(self) -> bool:
    """Check if Strapi API is available."""
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            res = await client.get(urljoin(self.base_url, "_health"))
            res.raise_for_status()
            return True
    except Exception:
        return False

StrapiClient

Bases: StrapiClientBase

REST API client for Strapi.

Source code in src/strapi_client/strapi_client.py 
 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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
class StrapiClient(StrapiClientBase):
    """REST API client for Strapi."""

    _client: httpx.Client | None = None

    def __enter__(self):
        self.open()
        return self

    def __exit__(self, exc_type, exc, tb):
        self.close()

    def open(self) -> httpx.Client:
        # Fallback to creating a client if not used in a context manager.
        if self._client is None:
            self._client = httpx.Client(timeout=self.timeout)
        return self._client

    def close(self):
        if self._client is not None:
            self._client.close()
            self._client = None

    @property
    def client(self) -> httpx.Client:
        if not self._client:
            raise RuntimeError("Client is not initialized.")
        return self._client

    def authorize(
        self,
        identifier: str,
        password: str,
    ) -> None:
        """Get auth token using identifier and password."""
        res = self.send_post_request(
            "auth/local", json=AuthPayload(identifier=identifier, password=password).model_dump(), use_auth=False
        )
        self._token = AuthResponse.model_validate(res.json()).jwt

    def get_single_document(self, single_api_id: str) -> DocumentResponse:
        """Get document of single type."""
        res = self.send_get_request(single_api_id)
        return DocumentResponse.model_validate(res.json())

    def get_document(
        self,
        plural_api_id: str,
        document_id: str,
        populate: list[str] | dict[str, Any] | str | None = None,
        fields: list[str] | None = None,
        locale: str | None = None,
    ) -> DocumentResponse:
        """Get document by document id."""
        params = ApiParameters(populate=populate, fields=fields, locale=locale)
        res = self.send_get_request(f"{plural_api_id}/{document_id}", params=params.stringify())
        return DocumentResponse.model_validate(res.json())

    def get_documents(
        self,
        plural_api_id: str,
        sort: list[str] | None = None,
        filters: dict[str, Any] | None = None,
        populate: list[str] | dict[str, Any] | str | None = None,
        fields: list[str] | None = None,
        publication_state: str | None = None,
        locale: str | None = None,
        start: int | None = None,
        page: int | None = None,
        batch_size: int = 25,
        with_count: bool = True,
    ) -> DocumentsResponse:
        """Get list of documents. By default, operates in batch mode to get all documents automatically."""
        params = ApiParameters(
            sort=sort,
            filters=filters,
            populate=populate,
            fields=fields,
            page=page,
            page_size=batch_size,
            start=start,
            limit=batch_size,
            publication_state=publication_state,
            locale=locale,
        )
        if params.page is not None or params.start is not None:  # Get specific page/batch
            res = self.send_get_request(plural_api_id, params=params.stringify())
            return DocumentsResponse.model_validate(res.json())
        else:  # Get all records
            params.start = 0
            params.with_count = True
            res = self.send_get_request(plural_api_id, params=params.stringify())
            res_page = DocumentsResponse.model_validate(res.json())
            start_list = [i for i in range(batch_size, res_page.meta.get_total_count(), batch_size)]
            all_data = res_page
            for cur_start in start_list:
                params.start = cur_start
                params.with_count = with_count
                res = self.send_get_request(plural_api_id, params=params.stringify())
                res_page = DocumentsResponse.model_validate(res.json())
                all_data.data += res_page.data
                all_data.meta = res_page.meta
            return all_data

    def create_or_update_single_document(
        self, single_api_id: str, data: dict[str, Any] | BaseModel
    ) -> DocumentResponse:
        """Create or update single type document."""
        res = self.send_put_request(single_api_id, body={"data": serialize_document_data(data)})
        return DocumentResponse.model_validate(res.json())

    def create_document(self, plural_api_id: str, data: dict[str, Any] | BaseModel) -> DocumentResponse:
        """Create new document."""
        res = self.send_post_request(
            plural_api_id,
            json={"data": serialize_document_data(data)},
        )
        return DocumentResponse.model_validate(res.json())

    def update_document(
        self, plural_api_id: str, document_id: str, data: dict[str, Any] | BaseModel
    ) -> DocumentResponse:
        """Update document fields."""
        res = self.send_put_request(
            f"{plural_api_id}/{document_id}",
            body={"data": serialize_document_data(data)},
        )
        return DocumentResponse.model_validate(res.json())

    def delete_single_document(self, single_api_id: str) -> None:
        """Delete single type document."""
        self.send_delete_request(single_api_id)

    def delete_document(self, plural_api_id: str, document_id: str) -> None:
        """Delete document by document id."""
        self.send_delete_request(f"{plural_api_id}/{document_id}")

    def send_get_request(
        self,
        route: str,
        params: dict[str, Any] | str | None = None,
        use_auth: bool = True,
    ) -> httpx.Response:
        """Send GET request to custom endpoint."""
        res = self.client.get(
            url=urljoin(self.api_url, route), params=params, headers=self._auth_header if use_auth else None
        )
        self._check_response(res, "Unable to send GET request")
        return res

    def send_put_request(
        self,
        route: str,
        body: dict[str, Any] | None = None,
        params: dict[str, Any] | str | None = None,
        use_auth: bool = True,
    ) -> httpx.Response:
        """Send PUT request to custom endpoint."""
        res = self.client.put(
            url=urljoin(self.api_url, route), json=body, params=params, headers=self._auth_header if use_auth else None
        )
        self._check_response(res, "Unable to send PUT request")
        return res

    def send_post_request(
        self,
        route: str,
        json: dict[str, Any] | None = None,
        params: dict[str, Any] | str | None = None,
        data: dict[str, Any] | None = None,
        files: list | None = None,
        use_auth: bool = True,
    ) -> httpx.Response:
        """Send POST request to custom endpoint."""
        res = self.client.post(
            url=urljoin(self.api_url, route),
            json=json,
            params=params,
            data=data,
            files=files,
            headers=self._auth_header if use_auth else None,
        )
        self._check_response(res, "Unable to send POST request")
        return res

    def send_delete_request(self, route: str, use_auth: bool = True) -> httpx.Response:
        """Send DELETE request to custom endpoint."""
        res = self.client.delete(url=urljoin(self.api_url, route), headers=self._auth_header if use_auth else None)
        self._check_response(res, "Unable to send DELETE request")
        return res

    def upload_files(
        self,
        files: list[Path | str] | dict[str, BytesIO | bytes | bytearray | memoryview],
        content_type_id: str | None = None,
        document_id: int | str | None = None,
        field: str | None = None,
    ) -> list[MediaImageDocument]:
        """Upload a list of files."""
        file_payloads = FilePayload.list_from_files(files)
        data: dict[str, Any] = {}
        if content_type_id and document_id and field:
            data = {"ref": content_type_id, "refId": document_id, "field": field}
        res = self.send_post_request("upload", data=data, files=[fp.to_files_tuple() for fp in file_payloads])
        self._check_response(res, "Unable to send POST request")
        return [MediaImageDocument.model_validate(d) for d in (res.json() or [])]

    def upload_file(
        self,
        file: Path | str | dict[str, BytesIO | bytes | bytearray | memoryview],
        content_type_id: str | None = None,
        document_id: int | str | None = None,
        field: str | None = None,
    ) -> MediaImageDocument:
        """Upload a list of files."""
        if isinstance(file, dict):
            if len(file) > 1:
                raise ValueError("One file must be provided in binary dict")
            files: list[Path | str] | dict[str, BytesIO | bytes | bytearray | memoryview] = file
        else:
            files = [file]
        result = self.upload_files(
            files=files,
            content_type_id=content_type_id,
            document_id=document_id,
            field=field,
        )
        if not result or len(result) != 1:
            raise ValueError("One and only one result is expected after operation")
        return result[0]

    def get_uploaded_files(self, filters: dict | None = None) -> list[dict[str, Any]]:
        """Get uploaded files."""
        params = ApiParameters(filters=filters)
        res = self.send_get_request("upload/files", params=params.stringify())
        return res.json()

    def check_health(self) -> bool:
        """Check if Strapi API is available."""
        try:
            with httpx.Client(timeout=5.0) as client:
                res = client.get(urljoin(self.base_url, "_health"))
                res.raise_for_status()
                return True
        except Exception:
            return False

authorize 

authorize(identifier, password)

Get auth token using identifier and password.

Source code in src/strapi_client/strapi_client.py 
45
46
47
48
49
50
51
52
53
54
def authorize(
    self,
    identifier: str,
    password: str,
) -> None:
    """Get auth token using identifier and password."""
    res = self.send_post_request(
        "auth/local", json=AuthPayload(identifier=identifier, password=password).model_dump(), use_auth=False
    )
    self._token = AuthResponse.model_validate(res.json()).jwt

get_single_document 

get_single_document(single_api_id)

Get document of single type.

Source code in src/strapi_client/strapi_client.py 
56
57
58
59
def get_single_document(self, single_api_id: str) -> DocumentResponse:
    """Get document of single type."""
    res = self.send_get_request(single_api_id)
    return DocumentResponse.model_validate(res.json())

get_document 

get_document(
    plural_api_id,
    document_id,
    populate=None,
    fields=None,
    locale=None,
)

Get document by document id.

Source code in src/strapi_client/strapi_client.py 
61
62
63
64
65
66
67
68
69
70
71
72
def get_document(
    self,
    plural_api_id: str,
    document_id: str,
    populate: list[str] | dict[str, Any] | str | None = None,
    fields: list[str] | None = None,
    locale: str | None = None,
) -> DocumentResponse:
    """Get document by document id."""
    params = ApiParameters(populate=populate, fields=fields, locale=locale)
    res = self.send_get_request(f"{plural_api_id}/{document_id}", params=params.stringify())
    return DocumentResponse.model_validate(res.json())

get_documents 

get_documents(
    plural_api_id,
    sort=None,
    filters=None,
    populate=None,
    fields=None,
    publication_state=None,
    locale=None,
    start=None,
    page=None,
    batch_size=25,
    with_count=True,
)

Get list of documents. By default, operates in batch mode to get all documents automatically.

Source code in src/strapi_client/strapi_client.py 
 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
def get_documents(
    self,
    plural_api_id: str,
    sort: list[str] | None = None,
    filters: dict[str, Any] | None = None,
    populate: list[str] | dict[str, Any] | str | None = None,
    fields: list[str] | None = None,
    publication_state: str | None = None,
    locale: str | None = None,
    start: int | None = None,
    page: int | None = None,
    batch_size: int = 25,
    with_count: bool = True,
) -> DocumentsResponse:
    """Get list of documents. By default, operates in batch mode to get all documents automatically."""
    params = ApiParameters(
        sort=sort,
        filters=filters,
        populate=populate,
        fields=fields,
        page=page,
        page_size=batch_size,
        start=start,
        limit=batch_size,
        publication_state=publication_state,
        locale=locale,
    )
    if params.page is not None or params.start is not None:  # Get specific page/batch
        res = self.send_get_request(plural_api_id, params=params.stringify())
        return DocumentsResponse.model_validate(res.json())
    else:  # Get all records
        params.start = 0
        params.with_count = True
        res = self.send_get_request(plural_api_id, params=params.stringify())
        res_page = DocumentsResponse.model_validate(res.json())
        start_list = [i for i in range(batch_size, res_page.meta.get_total_count(), batch_size)]
        all_data = res_page
        for cur_start in start_list:
            params.start = cur_start
            params.with_count = with_count
            res = self.send_get_request(plural_api_id, params=params.stringify())
            res_page = DocumentsResponse.model_validate(res.json())
            all_data.data += res_page.data
            all_data.meta = res_page.meta
        return all_data

create_or_update_single_document 

create_or_update_single_document(single_api_id, data)

Create or update single type document.

Source code in src/strapi_client/strapi_client.py 
120
121
122
123
124
125
def create_or_update_single_document(
    self, single_api_id: str, data: dict[str, Any] | BaseModel
) -> DocumentResponse:
    """Create or update single type document."""
    res = self.send_put_request(single_api_id, body={"data": serialize_document_data(data)})
    return DocumentResponse.model_validate(res.json())

create_document 

create_document(plural_api_id, data)

Create new document.

Source code in src/strapi_client/strapi_client.py 
127
128
129
130
131
132
133
def create_document(self, plural_api_id: str, data: dict[str, Any] | BaseModel) -> DocumentResponse:
    """Create new document."""
    res = self.send_post_request(
        plural_api_id,
        json={"data": serialize_document_data(data)},
    )
    return DocumentResponse.model_validate(res.json())

update_document 

update_document(plural_api_id, document_id, data)

Update document fields.

Source code in src/strapi_client/strapi_client.py 
135
136
137
138
139
140
141
142
143
def update_document(
    self, plural_api_id: str, document_id: str, data: dict[str, Any] | BaseModel
) -> DocumentResponse:
    """Update document fields."""
    res = self.send_put_request(
        f"{plural_api_id}/{document_id}",
        body={"data": serialize_document_data(data)},
    )
    return DocumentResponse.model_validate(res.json())

delete_single_document 

delete_single_document(single_api_id)

Delete single type document.

Source code in src/strapi_client/strapi_client.py 
145
146
147
def delete_single_document(self, single_api_id: str) -> None:
    """Delete single type document."""
    self.send_delete_request(single_api_id)

delete_document 

delete_document(plural_api_id, document_id)

Delete document by document id.

Source code in src/strapi_client/strapi_client.py 
149
150
151
def delete_document(self, plural_api_id: str, document_id: str) -> None:
    """Delete document by document id."""
    self.send_delete_request(f"{plural_api_id}/{document_id}")

send_get_request 

send_get_request(route, params=None, use_auth=True)

Send GET request to custom endpoint.

Source code in src/strapi_client/strapi_client.py 
153
154
155
156
157
158
159
160
161
162
163
164
def send_get_request(
    self,
    route: str,
    params: dict[str, Any] | str | None = None,
    use_auth: bool = True,
) -> httpx.Response:
    """Send GET request to custom endpoint."""
    res = self.client.get(
        url=urljoin(self.api_url, route), params=params, headers=self._auth_header if use_auth else None
    )
    self._check_response(res, "Unable to send GET request")
    return res

send_put_request 

send_put_request(
    route, body=None, params=None, use_auth=True
)

Send PUT request to custom endpoint.

Source code in src/strapi_client/strapi_client.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
def send_put_request(
    self,
    route: str,
    body: dict[str, Any] | None = None,
    params: dict[str, Any] | str | None = None,
    use_auth: bool = True,
) -> httpx.Response:
    """Send PUT request to custom endpoint."""
    res = self.client.put(
        url=urljoin(self.api_url, route), json=body, params=params, headers=self._auth_header if use_auth else None
    )
    self._check_response(res, "Unable to send PUT request")
    return res

send_post_request 

send_post_request(
    route,
    json=None,
    params=None,
    data=None,
    files=None,
    use_auth=True,
)

Send POST request to custom endpoint.

Source code in src/strapi_client/strapi_client.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def send_post_request(
    self,
    route: str,
    json: dict[str, Any] | None = None,
    params: dict[str, Any] | str | None = None,
    data: dict[str, Any] | None = None,
    files: list | None = None,
    use_auth: bool = True,
) -> httpx.Response:
    """Send POST request to custom endpoint."""
    res = self.client.post(
        url=urljoin(self.api_url, route),
        json=json,
        params=params,
        data=data,
        files=files,
        headers=self._auth_header if use_auth else None,
    )
    self._check_response(res, "Unable to send POST request")
    return res

send_delete_request 

send_delete_request(route, use_auth=True)

Send DELETE request to custom endpoint.

Source code in src/strapi_client/strapi_client.py 
201
202
203
204
205
def send_delete_request(self, route: str, use_auth: bool = True) -> httpx.Response:
    """Send DELETE request to custom endpoint."""
    res = self.client.delete(url=urljoin(self.api_url, route), headers=self._auth_header if use_auth else None)
    self._check_response(res, "Unable to send DELETE request")
    return res

upload_files 

upload_files(
    files,
    content_type_id=None,
    document_id=None,
    field=None,
)

Upload a list of files.

Source code in src/strapi_client/strapi_client.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def upload_files(
    self,
    files: list[Path | str] | dict[str, BytesIO | bytes | bytearray | memoryview],
    content_type_id: str | None = None,
    document_id: int | str | None = None,
    field: str | None = None,
) -> list[MediaImageDocument]:
    """Upload a list of files."""
    file_payloads = FilePayload.list_from_files(files)
    data: dict[str, Any] = {}
    if content_type_id and document_id and field:
        data = {"ref": content_type_id, "refId": document_id, "field": field}
    res = self.send_post_request("upload", data=data, files=[fp.to_files_tuple() for fp in file_payloads])
    self._check_response(res, "Unable to send POST request")
    return [MediaImageDocument.model_validate(d) for d in (res.json() or [])]

upload_file 

upload_file(
    file, content_type_id=None, document_id=None, field=None
)

Upload a list of files.

Source code in src/strapi_client/strapi_client.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def upload_file(
    self,
    file: Path | str | dict[str, BytesIO | bytes | bytearray | memoryview],
    content_type_id: str | None = None,
    document_id: int | str | None = None,
    field: str | None = None,
) -> MediaImageDocument:
    """Upload a list of files."""
    if isinstance(file, dict):
        if len(file) > 1:
            raise ValueError("One file must be provided in binary dict")
        files: list[Path | str] | dict[str, BytesIO | bytes | bytearray | memoryview] = file
    else:
        files = [file]
    result = self.upload_files(
        files=files,
        content_type_id=content_type_id,
        document_id=document_id,
        field=field,
    )
    if not result or len(result) != 1:
        raise ValueError("One and only one result is expected after operation")
    return result[0]

get_uploaded_files 

get_uploaded_files(filters=None)

Get uploaded files.

Source code in src/strapi_client/strapi_client.py 
247
248
249
250
251
def get_uploaded_files(self, filters: dict | None = None) -> list[dict[str, Any]]:
    """Get uploaded files."""
    params = ApiParameters(filters=filters)
    res = self.send_get_request("upload/files", params=params.stringify())
    return res.json()

check_health 

check_health()

Check if Strapi API is available.

Source code in src/strapi_client/strapi_client.py 
253
254
255
256
257
258
259
260
261
def check_health(self) -> bool:
    """Check if Strapi API is available."""
    try:
        with httpx.Client(timeout=5.0) as client:
            res = client.get(urljoin(self.base_url, "_health"))
            res.raise_for_status()
            return True
    except Exception:
        return False

BaseDocument pydantic-model

Bases: BasePopulatable

Strapi document with standard fields.

Fields:

  • id (int)
  • document_id (str)
  • created_at (datetime)
  • updated_at (datetime)
  • published_at (datetime)
Source code in src/strapi_client/models/base_document.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class BaseDocument(BasePopulatable):
    """Strapi document with standard fields."""

    id: int
    document_id: str = Field(alias="documentId")
    created_at: datetime.datetime = Field(alias="createdAt")
    updated_at: datetime.datetime = Field(alias="updatedAt")
    published_at: datetime.datetime = Field(alias="publishedAt")

    @classmethod
    def from_scalar_response(cls, response: DocumentResponse) -> Self:
        return cls.model_validate(response.data)

    @classmethod
    def from_list_response(cls, response: DocumentsResponse) -> list[Self]:
        return [cls.model_validate(d) for d in response.data]

    @classmethod
    def first_from_list_response(cls, response: DocumentsResponse) -> Self | None:
        return cls.model_validate(response.data[0]) if len(response.data) > 0 else None

BaseComponent pydantic-model

Bases: BasePopulatable

Strapi component.

Source code in src/strapi_client/models/base_component.py 
4
5
class BaseComponent(BasePopulatable):
    """Strapi component."""

MediaImageDocument pydantic-model

Bases: BaseDocument

Fields:

  • id (int)
  • document_id (str)
  • created_at (datetime)
  • updated_at (datetime)
  • published_at (datetime)
  • name (str)
  • alternative_text (str | None)
  • caption (str | None)
  • width (int | None)
  • height (int | None)
  • formats (MediaImageFormats | None)
  • hash (str)
  • ext (str)
  • mime (str)
  • size (float)
  • url (str)
  • preview_url (str | None)
  • provider (str)
  • provider_metadata (dict[str, Any] | None)
Source code in src/strapi_client/models/media_image_document.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class MediaImageDocument(BaseDocument):
    name: str
    alternative_text: str | None = Field(default=None, alias="alternativeText")
    caption: str | None = None
    width: int | None
    height: int | None
    formats: MediaImageFormats | None
    hash: str
    ext: str
    mime: str
    size: float = Field(description="Size in kilobytes")
    url: str
    preview_url: str | None = Field(default=None, alias="previewUrl")
    provider: str
    provider_metadata: dict[str, Any] | None = None

    @property
    def largest_format(self) -> MediaImageFormatVariant | None:
        return self.formats.largest if self.formats else None

size pydantic-field 

size

Size in kilobytes

SmartDocument pydantic-model

Bases: BaseDocument

Fully automated ORM class for Strapi documents.

Fields:

  • id (int)
  • document_id (str)
  • created_at (datetime)
  • updated_at (datetime)
  • published_at (datetime)
Source code in src/strapi_client/models/smart_document.py 
 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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
class SmartDocument(BaseDocument):
    """Fully automated ORM class for Strapi documents."""

    __singular_api_id__: ClassVar[str]
    __plural_api_id__: ClassVar[str]
    __content_type_id__: ClassVar[str]
    __managed_fields__: ClassVar[set[str]] = {
        "id",
        "document_id",
        "created_at",
        "updated_at",
        "published_at",
        "__singular_api_id__",
        "__plural_api_id__",
        "__content_type_id__",
        "__managed_fields__",
    }

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        super().__pydantic_init_subclass__(**kwargs)
        if not hasattr(cls, "__singular_api_id__"):
            setattr(cls, "__singular_api_id__", re.sub(r"(?<!^)(?=[A-Z])", "-", cls.__name__).lower())
        if not hasattr(cls, "__plural_api_id__"):
            setattr(cls, "__plural_api_id__", f"{getattr(cls, '__singular_api_id__')}s")
        if not hasattr(cls, "__content_type_id__"):
            setattr(
                cls,
                "__content_type_id__",
                f"api::{getattr(cls, '__singular_api_id__')}.{getattr(cls, '__singular_api_id__')}",
            )

    @classmethod
    async def get_document(
        cls,
        client: StrapiClientAsync,
        document_id: str,
    ) -> Self:
        """Get document by document id."""
        fields, populate = get_model_fields_and_population(cls)
        response = await client.get_document(
            plural_api_id=cls.__plural_api_id__,
            document_id=document_id,
            fields=fields,
            populate=populate,
        )
        return cls.from_scalar_response(response)

    @classmethod
    async def get_documents(
        cls,
        client: StrapiClientAsync,
        sort: list[str] | None = None,
        filters: dict[str, Any] | None = None,
        publication_state: str | None = None,
        locale: str | None = None,
        start: int | None = None,
        page: int | None = None,
        limit: int = 100,
        with_count: bool = True,
    ) -> list[Self]:
        """Get list of documents."""
        fields, populate = get_model_fields_and_population(cls)
        response = await client.get_documents(
            plural_api_id=cls.__plural_api_id__,
            sort=sort or ["id"],
            filters=filters,
            populate=populate,
            fields=fields,
            publication_state=publication_state,
            locale=locale,
            page=page,
            start=start,
            batch_size=limit,
            with_count=with_count,
        )
        return cls.from_list_response(response)

    @classmethod
    async def get_documents_with_meta(
        cls,
        client: StrapiClientAsync,
        sort: list[str] | None = None,
        filters: dict[str, Any] | None = None,
        publication_state: str | None = None,
        locale: str | None = None,
        start: int | None = None,
        page: int | None = None,
        limit: int = 100,
        with_count: bool = True,
    ) -> tuple[list[Self], ResponseMeta]:
        """Get list of documents."""
        fields, populate = get_model_fields_and_population(cls)
        response = await client.get_documents(
            plural_api_id=cls.__plural_api_id__,
            sort=sort,
            filters=filters,
            populate=populate,
            fields=fields,
            publication_state=publication_state,
            locale=locale,
            page=page,
            start=start,
            batch_size=limit,
            with_count=with_count,
        )
        return cls.from_list_response(response), response.meta

    @classmethod
    async def get_first_document(
        cls,
        client: StrapiClientAsync,
        sort: list[str] | None = None,
        filters: dict[str, Any] | None = None,
        publication_state: str | None = None,
        locale: str | None = None,
    ) -> Self | None:
        """First documents if available."""
        fields, populate = get_model_fields_and_population(cls)
        response = await client.get_documents(
            plural_api_id=cls.__plural_api_id__,
            sort=sort,
            filters=filters,
            populate=populate,
            fields=fields,
            publication_state=publication_state,
            locale=locale,
            start=0,
            batch_size=1,
            with_count=False,
        )
        return cls.first_from_list_response(response)

    @classmethod
    async def create_document(
        cls,
        client: StrapiClientAsync,
        data: dict[str, Any] | BaseModel,
    ) -> Self:
        """Create a new document."""
        response = await client.create_document(
            plural_api_id=cls.__plural_api_id__,
            data=serialize_document_data(data),
        )
        _, populate = get_model_fields_and_population(cls)
        if not populate:
            return cls.from_scalar_response(response)
        else:
            result_document = BaseDocument.from_scalar_response(response)
            return await cls.get_document(client, result_document.document_id)

    def model_dump_data(self, exclude_managed_fields: bool = False, json_mode: bool = False) -> dict[str, Any]:
        """
        Create a dictionary representation of the document.

        Args:
            exclude_managed_fields: If True, exclude fields listed in __managed_fields__
            json_mode: If True, serialize fields to JSON compatible types

        Returns:
            Dictionary representation of the document with nested BaseDocument instances
            replaced with their IDs
        """
        return get_model_data(self, exclude_managed_fields=exclude_managed_fields, json_mode=json_mode)

    def model_identical(self, data: dict[str, Any] | BaseModel, exclude_fields: list[str] | None = None) -> bool:
        data_dict = data.model_dump(by_alias=True) if isinstance(data, BaseModel) else data
        data_dict_filtered = {k: v for k, v in data_dict.items() if k not in (exclude_fields or [])}
        record_dict = self.model_dump_data(exclude_managed_fields=True)
        record_dict_filtered = {k: v for k, v in record_dict.items() if k in data_dict_filtered}
        return hash_model(record_dict_filtered) == hash_model(data_dict_filtered)

    async def update_document(
        self,
        client: StrapiClientAsync,
        data: dict[str, Any] | BaseModel,
        lazy_mode: bool = False,
        do_not_compare_fields: list[str] | None = None,
    ) -> Self:
        """Update existing document."""
        _, populate = get_model_fields_and_population(self.__class__)
        if not lazy_mode and do_not_compare_fields:
            warnings.warn("do_not_compare_fields argument works only in lazy mode")
        elif lazy_mode and self.model_identical(data=data, exclude_fields=do_not_compare_fields):
            return self
        response = await client.update_document(
            plural_api_id=self.__plural_api_id__,
            document_id=self.document_id,
            data=serialize_document_data(data),
        )
        if not populate:
            result_document = self.__class__.from_scalar_response(response)
            self.__dict__.update(result_document.__dict__)
            return self
        else:
            return await self.refresh_document(client)

    async def lazy_update_document(
        self,
        client: StrapiClientAsync,
        data: dict[str, Any] | BaseModel,
        do_not_compare_fields: list[str] | None = None,
    ) -> bool:
        """Lazy update existing document fields without record synchronization."""
        if self.model_identical(data=data, exclude_fields=do_not_compare_fields):
            return False
        else:
            await client.update_document(
                plural_api_id=self.__plural_api_id__,
                document_id=self.document_id,
                data=serialize_document_data(data),
            )
            return True

    async def update_relations(
        self,
        client: StrapiClientAsync,
        field: str,
        relations: list[BaseDocument] | None = None,
        connect: list[BaseDocument] | None = None,
        disconnect: list[BaseDocument] | None = None,
    ) -> Self:
        """Update field relations."""
        if not relations and not connect and not disconnect:
            raise ValueError("At least one of relations, connect or disconnect should be provided")
        if relations and (connect or disconnect):
            raise ValueError("relations argument does not work with connect or disconnect arguments")
        if relations:
            data = {field: {"set": [d.document_id for d in relations]}}
        else:
            data = {
                field: {
                    "connect": [d.document_id for d in connect or {}],
                    "disconnect": [d.document_id for d in disconnect or {}],
                }
            }
        await client.update_document(
            plural_api_id=self.__plural_api_id__,
            document_id=self.document_id,
            data=data,
        )
        return await self.refresh_document(client)

    async def refresh_document(self, client: StrapiClientAsync) -> Self:
        """Refresh the document with the latest data from Strapi."""
        fields, populate = get_model_fields_and_population(self.__class__)
        response = await client.get_document(
            plural_api_id=self.__plural_api_id__,
            document_id=self.document_id,
            fields=fields,
            populate=populate,
        )
        document = self.from_scalar_response(response)
        self.__dict__.update(document.__dict__)
        return self

    async def delete_document(self, client: StrapiClientAsync) -> None:
        """Delete the document."""
        await client.delete_document(
            plural_api_id=self.__plural_api_id__,
            document_id=self.document_id,
        )

    async def upload_file(
        self,
        client: StrapiClientAsync,
        file: Path | str | dict[str, BytesIO | bytes | bytearray | memoryview],
        field: str,
    ) -> Self:
        """Upload a file to the document's field."""
        if isinstance(file, dict):
            file_data: list[Path | str] | dict[str, BytesIO | bytes | bytearray | memoryview] = file
        else:
            file_data = [file]
        await client.upload_files(
            files=file_data,
            content_type_id=self.__content_type_id__,
            document_id=self.id,
            field=field,
        )
        return await self.refresh_document(client)

get_document async classmethod 

get_document(client, document_id)

Get document by document id.

Source code in src/strapi_client/models/smart_document.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@classmethod
async def get_document(
    cls,
    client: StrapiClientAsync,
    document_id: str,
) -> Self:
    """Get document by document id."""
    fields, populate = get_model_fields_and_population(cls)
    response = await client.get_document(
        plural_api_id=cls.__plural_api_id__,
        document_id=document_id,
        fields=fields,
        populate=populate,
    )
    return cls.from_scalar_response(response)

get_documents async classmethod 

get_documents(
    client,
    sort=None,
    filters=None,
    publication_state=None,
    locale=None,
    start=None,
    page=None,
    limit=100,
    with_count=True,
)

Get list of documents.

Source code in src/strapi_client/models/smart_document.py 
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
@classmethod
async def get_documents(
    cls,
    client: StrapiClientAsync,
    sort: list[str] | None = None,
    filters: dict[str, Any] | None = None,
    publication_state: str | None = None,
    locale: str | None = None,
    start: int | None = None,
    page: int | None = None,
    limit: int = 100,
    with_count: bool = True,
) -> list[Self]:
    """Get list of documents."""
    fields, populate = get_model_fields_and_population(cls)
    response = await client.get_documents(
        plural_api_id=cls.__plural_api_id__,
        sort=sort or ["id"],
        filters=filters,
        populate=populate,
        fields=fields,
        publication_state=publication_state,
        locale=locale,
        page=page,
        start=start,
        batch_size=limit,
        with_count=with_count,
    )
    return cls.from_list_response(response)

get_documents_with_meta async classmethod 

get_documents_with_meta(
    client,
    sort=None,
    filters=None,
    publication_state=None,
    locale=None,
    start=None,
    page=None,
    limit=100,
    with_count=True,
)

Get list of documents.

Source code in src/strapi_client/models/smart_document.py 
 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
@classmethod
async def get_documents_with_meta(
    cls,
    client: StrapiClientAsync,
    sort: list[str] | None = None,
    filters: dict[str, Any] | None = None,
    publication_state: str | None = None,
    locale: str | None = None,
    start: int | None = None,
    page: int | None = None,
    limit: int = 100,
    with_count: bool = True,
) -> tuple[list[Self], ResponseMeta]:
    """Get list of documents."""
    fields, populate = get_model_fields_and_population(cls)
    response = await client.get_documents(
        plural_api_id=cls.__plural_api_id__,
        sort=sort,
        filters=filters,
        populate=populate,
        fields=fields,
        publication_state=publication_state,
        locale=locale,
        page=page,
        start=start,
        batch_size=limit,
        with_count=with_count,
    )
    return cls.from_list_response(response), response.meta

get_first_document async classmethod 

get_first_document(
    client,
    sort=None,
    filters=None,
    publication_state=None,
    locale=None,
)

First documents if available.

Source code in src/strapi_client/models/smart_document.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@classmethod
async def get_first_document(
    cls,
    client: StrapiClientAsync,
    sort: list[str] | None = None,
    filters: dict[str, Any] | None = None,
    publication_state: str | None = None,
    locale: str | None = None,
) -> Self | None:
    """First documents if available."""
    fields, populate = get_model_fields_and_population(cls)
    response = await client.get_documents(
        plural_api_id=cls.__plural_api_id__,
        sort=sort,
        filters=filters,
        populate=populate,
        fields=fields,
        publication_state=publication_state,
        locale=locale,
        start=0,
        batch_size=1,
        with_count=False,
    )
    return cls.first_from_list_response(response)

create_document async classmethod 

create_document(client, data)

Create a new document.

Source code in src/strapi_client/models/smart_document.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@classmethod
async def create_document(
    cls,
    client: StrapiClientAsync,
    data: dict[str, Any] | BaseModel,
) -> Self:
    """Create a new document."""
    response = await client.create_document(
        plural_api_id=cls.__plural_api_id__,
        data=serialize_document_data(data),
    )
    _, populate = get_model_fields_and_population(cls)
    if not populate:
        return cls.from_scalar_response(response)
    else:
        result_document = BaseDocument.from_scalar_response(response)
        return await cls.get_document(client, result_document.document_id)

model_dump_data 

model_dump_data(
    exclude_managed_fields=False, json_mode=False
)

Create a dictionary representation of the document.

Parameters:
  • exclude_managed_fields (bool, default: False ) –

    If True, exclude fields listed in managed_fields

  • json_mode (bool, default: False ) –

    If True, serialize fields to JSON compatible types
Returns:
  • dict[str, Any] –

    Dictionary representation of the document with nested BaseDocument instances

  • dict[str, Any] –

    replaced with their IDs
Source code in src/strapi_client/models/smart_document.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
def model_dump_data(self, exclude_managed_fields: bool = False, json_mode: bool = False) -> dict[str, Any]:
    """
    Create a dictionary representation of the document.

    Args:
        exclude_managed_fields: If True, exclude fields listed in __managed_fields__
        json_mode: If True, serialize fields to JSON compatible types

    Returns:
        Dictionary representation of the document with nested BaseDocument instances
        replaced with their IDs
    """
    return get_model_data(self, exclude_managed_fields=exclude_managed_fields, json_mode=json_mode)

update_document async 

update_document(
    client,
    data,
    lazy_mode=False,
    do_not_compare_fields=None,
)

Update existing document.

Source code in src/strapi_client/models/smart_document.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
async def update_document(
    self,
    client: StrapiClientAsync,
    data: dict[str, Any] | BaseModel,
    lazy_mode: bool = False,
    do_not_compare_fields: list[str] | None = None,
) -> Self:
    """Update existing document."""
    _, populate = get_model_fields_and_population(self.__class__)
    if not lazy_mode and do_not_compare_fields:
        warnings.warn("do_not_compare_fields argument works only in lazy mode")
    elif lazy_mode and self.model_identical(data=data, exclude_fields=do_not_compare_fields):
        return self
    response = await client.update_document(
        plural_api_id=self.__plural_api_id__,
        document_id=self.document_id,
        data=serialize_document_data(data),
    )
    if not populate:
        result_document = self.__class__.from_scalar_response(response)
        self.__dict__.update(result_document.__dict__)
        return self
    else:
        return await self.refresh_document(client)

lazy_update_document async 

lazy_update_document(
    client, data, do_not_compare_fields=None
)

Lazy update existing document fields without record synchronization.

Source code in src/strapi_client/models/smart_document.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
async def lazy_update_document(
    self,
    client: StrapiClientAsync,
    data: dict[str, Any] | BaseModel,
    do_not_compare_fields: list[str] | None = None,
) -> bool:
    """Lazy update existing document fields without record synchronization."""
    if self.model_identical(data=data, exclude_fields=do_not_compare_fields):
        return False
    else:
        await client.update_document(
            plural_api_id=self.__plural_api_id__,
            document_id=self.document_id,
            data=serialize_document_data(data),
        )
        return True

update_relations async 

update_relations(
    client,
    field,
    relations=None,
    connect=None,
    disconnect=None,
)

Update field relations.

Source code in src/strapi_client/models/smart_document.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
async def update_relations(
    self,
    client: StrapiClientAsync,
    field: str,
    relations: list[BaseDocument] | None = None,
    connect: list[BaseDocument] | None = None,
    disconnect: list[BaseDocument] | None = None,
) -> Self:
    """Update field relations."""
    if not relations and not connect and not disconnect:
        raise ValueError("At least one of relations, connect or disconnect should be provided")
    if relations and (connect or disconnect):
        raise ValueError("relations argument does not work with connect or disconnect arguments")
    if relations:
        data = {field: {"set": [d.document_id for d in relations]}}
    else:
        data = {
            field: {
                "connect": [d.document_id for d in connect or {}],
                "disconnect": [d.document_id for d in disconnect or {}],
            }
        }
    await client.update_document(
        plural_api_id=self.__plural_api_id__,
        document_id=self.document_id,
        data=data,
    )
    return await self.refresh_document(client)

refresh_document async 

refresh_document(client)

Refresh the document with the latest data from Strapi.

Source code in src/strapi_client/models/smart_document.py 
258
259
260
261
262
263
264
265
266
267
268
269
async def refresh_document(self, client: StrapiClientAsync) -> Self:
    """Refresh the document with the latest data from Strapi."""
    fields, populate = get_model_fields_and_population(self.__class__)
    response = await client.get_document(
        plural_api_id=self.__plural_api_id__,
        document_id=self.document_id,
        fields=fields,
        populate=populate,
    )
    document = self.from_scalar_response(response)
    self.__dict__.update(document.__dict__)
    return self

delete_document async 

delete_document(client)

Delete the document.

Source code in src/strapi_client/models/smart_document.py 
271
272
273
274
275
276
async def delete_document(self, client: StrapiClientAsync) -> None:
    """Delete the document."""
    await client.delete_document(
        plural_api_id=self.__plural_api_id__,
        document_id=self.document_id,
    )

upload_file async 

upload_file(client, file, field)

Upload a file to the document's field.

Source code in src/strapi_client/models/smart_document.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
async def upload_file(
    self,
    client: StrapiClientAsync,
    file: Path | str | dict[str, BytesIO | bytes | bytearray | memoryview],
    field: str,
) -> Self:
    """Upload a file to the document's field."""
    if isinstance(file, dict):
        file_data: list[Path | str] | dict[str, BytesIO | bytes | bytearray | memoryview] = file
    else:
        file_data = [file]
    await client.upload_files(
        files=file_data,
        content_type_id=self.__content_type_id__,
        document_id=self.id,
        field=field,
    )
    return await self.refresh_document(client)

SingleSmartDocument pydantic-model

Bases: BaseDocument

Fully automated ORM class for single Strapi document.

Fields:

  • id (int)
  • document_id (str)
  • created_at (datetime)
  • updated_at (datetime)
  • published_at (datetime)
Source code in src/strapi_client/models/single_smart_document.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class SingleSmartDocument(BaseDocument):
    """Fully automated ORM class for single Strapi document."""

    __single_api_id__: ClassVar[str]

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        super().__pydantic_init_subclass__(**kwargs)
        if "__single_api_id__" not in cls.__dict__:
            name = re.sub(r"(?<!^)(?=[A-Z])", "-", cls.__name__).lower()
            cls.__single_api_id__ = name

    @classmethod
    async def get_document(
        cls,
        client: StrapiClientAsync,
    ) -> Self:
        """Get single document by single api id."""
        response = await client.get_single_document(single_api_id=cls.__single_api_id__)
        return cls.from_scalar_response(response)

get_document async classmethod 

get_document(client)

Get single document by single api id.

Source code in src/strapi_client/models/single_smart_document.py 
20
21
22
23
24
25
26
27
@classmethod
async def get_document(
    cls,
    client: StrapiClientAsync,
) -> Self:
    """Get single document by single api id."""
    response = await client.get_single_document(single_api_id=cls.__single_api_id__)
    return cls.from_scalar_response(response)

ActiveDocument pydantic-model

Bases: BaseModel

Experimental ORM class for Strapi document.

Fields:

  • id (int | None)
  • document_id (str | None)
  • created_at (datetime | None)
  • updated_at (datetime | None)
  • published_at (datetime | None)
Source code in src/strapi_client/models/active_document.py 
 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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
class ActiveDocument(BaseModel):
    """Experimental ORM class for Strapi document."""

    __plural_api_id__: ClassVar[str]
    __managed_fields__: ClassVar[set[str]] = {
        "id",
        "document_id",
        "created_at",
        "updated_at",
        "published_at",
        "__plural_api_id__",
        "__managed_fields__",
    }
    _relations_populated: bool = PrivateAttr(default=False)
    id: int | None = DocumentField(default=None, unique=True)
    document_id: str | None = DocumentField(default=None, alias="documentId", unique=True)
    created_at: datetime.datetime | None = Field(default=None, alias="createdAt")
    updated_at: datetime.datetime | None = Field(default=None, alias="updatedAt")
    published_at: datetime.datetime | None = Field(default=None, alias="publishedAt")

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        super().__pydantic_init_subclass__(**kwargs)
        if "__plural_api_id__" not in cls.__dict__:
            cls.__plural_api_id__ = cls.__name__.lower() + "s"

    @classmethod
    async def get_document(
        cls,
        client: StrapiClientAsync,
        document_id: str,
        populate_all: bool = True,
    ) -> Self:
        """Get document by document id."""
        response = await client.get_document(
            plural_api_id=cls.__plural_api_id__,
            document_id=document_id,
            populate=list(cls._get_relation_fields()) if populate_all else None,
        )
        return cls.model_validate(response.data)

    @classmethod
    async def get_documents(
        cls,
        client: StrapiClientAsync,
        populate_all: bool = True,
        sort: list[str] | None = None,
        filters: dict[str, Any] | None = None,
        publication_state: str | None = None,
        locale: str | None = None,
        start: int | None = 0,
        page: int | None = None,
        limit: int = 25,
        with_count: bool = True,
    ) -> list[Self]:
        """Get list of documents."""
        response = await client.get_documents(
            plural_api_id=cls.__plural_api_id__,
            sort=sort,
            filters=filters,
            populate=list(cls._get_relation_fields()) if populate_all else None,
            fields=list(cls._get_document_fields(with_relations=False)),
            publication_state=publication_state,
            locale=locale,
            page=page,
            start=start,
            batch_size=limit,
            with_count=with_count,
        )
        documents = [cls.model_validate(document) for document in response.data]
        if populate_all:
            for document in documents:
                document.set_relations_populated(populate_all)
        return documents

    async def create_document(self, client: StrapiClientAsync) -> Self:
        """Create new document from object."""
        response = await client.create_document(plural_api_id=self.__plural_api_id__, data=self.model_dump_variable())
        return self.model_validate(response.data)

    async def update_document(self, client: StrapiClientAsync) -> Self:
        """Update document fields from object."""
        if not self.document_id:
            raise RuntimeError("Document ID cannot be empty to update document")
        if not self.relations_populated():
            warnings.warn(
                "Some relations are not populated, so all relations will not be updated. Use refresh() method to populate relations."
            )
            # TODO: relations: set, connect, disconnect. Preserve hashing. Check warning
            data = self.model_dump_variable(exclude=self._get_relation_fields())
        else:
            data = self.model_dump_variable()
        response = await client.update_document(
            plural_api_id=self.__plural_api_id__, document_id=self.document_id, data=data
        )
        return self.model_validate(response.data)

    async def delete_document(self, client: StrapiClientAsync) -> None:
        """Delete document attached to object."""
        if not self.document_id:
            raise RuntimeError("Document ID cannot be empty to delete document")
        await client.delete_document(plural_api_id=self.__plural_api_id__, document_id=self.document_id)

    def relations_populated(self) -> bool:
        return self._relations_populated or not self._get_relation_fields()

    def set_relations_populated(self, value: bool) -> Self:
        self._relations_populated = value
        return self

    async def refresh(self, client: StrapiClientAsync, populate_all: bool = True) -> Self:
        """Refresh object with latest data from Strapi including relations."""
        if not self.document_id:
            raise RuntimeError("Document ID cannot be empty to refresh object")
        document = await self.__class__.get_document(client, self.document_id, populate_all=populate_all)
        for field in document.model_fields:
            setattr(self, field, getattr(document, field))

        self.set_relations_populated(populate_all)
        return self

    async def upsert_document(self, client: StrapiClientAsync) -> Self:
        """Create document or update fields."""
        keys: set[str] = self._unique_fields - self.__managed_fields__
        if not keys:
            raise RuntimeError("For upsert at least one model field should be declared as unique")
        model_dict = self.model_dump_variable()
        filters = {key: {"$eq": model_dict[key]} for key in keys}
        cur_response = await client.get_documents(
            plural_api_id=self.__plural_api_id__,
            filters=filters,
            fields=list(self._get_document_fields()),
            start=0,
            batch_size=1,
            with_count=True,
        )
        total_count: int = cur_response.meta.get_total_count()
        if total_count > 1:
            raise RuntimeError(f"Keys are ambiguous, found {total_count} records")
        elif total_count == 0:
            return await self.create_document(client)
        else:
            cur_document = self.model_validate(cur_response.data[0])
            if cur_document.model_hash() == self.model_hash():
                return cur_document
            else:
                self.id = cur_document.id
                self.document_id = cur_document.document_id
                return await self.update_document(client)

    def model_dump_variable(self, exclude: set[str] | None = None) -> dict[str, Any]:
        exclude = exclude or set()
        model_dict = self.model_dump(by_alias=True, exclude=self.__managed_fields__ | exclude)
        for rel in self._get_relation_fields():
            if model_dict.get(rel):
                model_dict[rel] = {"set": model_dict[rel]["documentId"]}
        return model_dict

    def model_hash(self) -> str:
        dumped_str = json.dumps(self.model_dump_variable(), sort_keys=True, default=str)
        return hashlib.sha256(dumped_str.encode("utf-8")).hexdigest()

    @property
    def _unique_fields(self) -> set[str]:
        return {f for f, info in self.model_fields.items() if any(m.get("unique", False) for m in info.metadata)}

    @classmethod
    def _get_document_fields(cls, with_relations: bool = True) -> set[str]:
        return {
            info.alias or f
            for f, info in cls.__pydantic_fields__.items()
            if with_relations or not any(m.get("relation", False) for m in info.metadata)
        }

    @classmethod
    def _get_relation_fields(cls) -> set[str]:
        return {
            info.alias or f
            for f, info in cls.__pydantic_fields__.items()
            if any(m.get("relation", False) for m in info.metadata)
        }

get_document async classmethod 

get_document(client, document_id, populate_all=True)

Get document by document id.

Source code in src/strapi_client/models/active_document.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@classmethod
async def get_document(
    cls,
    client: StrapiClientAsync,
    document_id: str,
    populate_all: bool = True,
) -> Self:
    """Get document by document id."""
    response = await client.get_document(
        plural_api_id=cls.__plural_api_id__,
        document_id=document_id,
        populate=list(cls._get_relation_fields()) if populate_all else None,
    )
    return cls.model_validate(response.data)

get_documents async classmethod 

get_documents(
    client,
    populate_all=True,
    sort=None,
    filters=None,
    publication_state=None,
    locale=None,
    start=0,
    page=None,
    limit=25,
    with_count=True,
)

Get list of documents.

Source code in src/strapi_client/models/active_document.py 
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
@classmethod
async def get_documents(
    cls,
    client: StrapiClientAsync,
    populate_all: bool = True,
    sort: list[str] | None = None,
    filters: dict[str, Any] | None = None,
    publication_state: str | None = None,
    locale: str | None = None,
    start: int | None = 0,
    page: int | None = None,
    limit: int = 25,
    with_count: bool = True,
) -> list[Self]:
    """Get list of documents."""
    response = await client.get_documents(
        plural_api_id=cls.__plural_api_id__,
        sort=sort,
        filters=filters,
        populate=list(cls._get_relation_fields()) if populate_all else None,
        fields=list(cls._get_document_fields(with_relations=False)),
        publication_state=publication_state,
        locale=locale,
        page=page,
        start=start,
        batch_size=limit,
        with_count=with_count,
    )
    documents = [cls.model_validate(document) for document in response.data]
    if populate_all:
        for document in documents:
            document.set_relations_populated(populate_all)
    return documents

create_document async 

create_document(client)

Create new document from object.

Source code in src/strapi_client/models/active_document.py 
92
93
94
95
async def create_document(self, client: StrapiClientAsync) -> Self:
    """Create new document from object."""
    response = await client.create_document(plural_api_id=self.__plural_api_id__, data=self.model_dump_variable())
    return self.model_validate(response.data)

update_document async 

update_document(client)

Update document fields from object.

Source code in src/strapi_client/models/active_document.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async def update_document(self, client: StrapiClientAsync) -> Self:
    """Update document fields from object."""
    if not self.document_id:
        raise RuntimeError("Document ID cannot be empty to update document")
    if not self.relations_populated():
        warnings.warn(
            "Some relations are not populated, so all relations will not be updated. Use refresh() method to populate relations."
        )
        # TODO: relations: set, connect, disconnect. Preserve hashing. Check warning
        data = self.model_dump_variable(exclude=self._get_relation_fields())
    else:
        data = self.model_dump_variable()
    response = await client.update_document(
        plural_api_id=self.__plural_api_id__, document_id=self.document_id, data=data
    )
    return self.model_validate(response.data)

delete_document async 

delete_document(client)

Delete document attached to object.

Source code in src/strapi_client/models/active_document.py 
114
115
116
117
118
async def delete_document(self, client: StrapiClientAsync) -> None:
    """Delete document attached to object."""
    if not self.document_id:
        raise RuntimeError("Document ID cannot be empty to delete document")
    await client.delete_document(plural_api_id=self.__plural_api_id__, document_id=self.document_id)

refresh async 

refresh(client, populate_all=True)

Refresh object with latest data from Strapi including relations.

Source code in src/strapi_client/models/active_document.py 
127
128
129
130
131
132
133
134
135
136
async def refresh(self, client: StrapiClientAsync, populate_all: bool = True) -> Self:
    """Refresh object with latest data from Strapi including relations."""
    if not self.document_id:
        raise RuntimeError("Document ID cannot be empty to refresh object")
    document = await self.__class__.get_document(client, self.document_id, populate_all=populate_all)
    for field in document.model_fields:
        setattr(self, field, getattr(document, field))

    self.set_relations_populated(populate_all)
    return self

upsert_document async 

upsert_document(client)

Create document or update fields.

Source code in src/strapi_client/models/active_document.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
async def upsert_document(self, client: StrapiClientAsync) -> Self:
    """Create document or update fields."""
    keys: set[str] = self._unique_fields - self.__managed_fields__
    if not keys:
        raise RuntimeError("For upsert at least one model field should be declared as unique")
    model_dict = self.model_dump_variable()
    filters = {key: {"$eq": model_dict[key]} for key in keys}
    cur_response = await client.get_documents(
        plural_api_id=self.__plural_api_id__,
        filters=filters,
        fields=list(self._get_document_fields()),
        start=0,
        batch_size=1,
        with_count=True,
    )
    total_count: int = cur_response.meta.get_total_count()
    if total_count > 1:
        raise RuntimeError(f"Keys are ambiguous, found {total_count} records")
    elif total_count == 0:
        return await self.create_document(client)
    else:
        cur_document = self.model_validate(cur_response.data[0])
        if cur_document.model_hash() == self.model_hash():
            return cur_document
        else:
            self.id = cur_document.id
            self.document_id = cur_document.document_id
            return await self.update_document(client)