Добавление Assistants API, возможность загружать файлы + добавление недостающих полей в pydantic модели

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "gigachat"
-version = "0.1.24"
+version = "0.1.26"
 description = "GigaChat. Python-library for GigaChain and LangChain"
 authors = ["Konstantin Krestnikov <[email protected]>", "Sergey Malyshev <[email protected]>"]
 license = "MIT"

src/gigachat/_types.py

@@ -0,0 +1,13 @@
+from typing import IO, Mapping, Optional, Tuple, Union
+
+FileContent = Union[IO[bytes], bytes, str]
+FileTypes = Union[
+    # file (or bytes)
+    FileContent,
+    # (filename, file (or bytes))
+    Tuple[Optional[str], FileContent],
+    # (filename, file (or bytes), content_type)
+    Tuple[Optional[str], FileContent, Optional[str]],
+    # (filename, file (or bytes), content_type, headers)
+    Tuple[Optional[str], FileContent, Optional[str], Mapping[str, str]],
+]

src/gigachat/api/assistants/get_assistants.py

@@ -0,0 +1,57 @@
+from http import HTTPStatus
+from typing import Any, Dict, Optional
+
+import httpx
+
+from gigachat.api.utils import build_headers
+from gigachat.exceptions import AuthenticationError, ResponseError
+from gigachat.models.assistants import Assistants
+
+
+def _get_kwargs(
+    *,
+    assistant_id: Optional[str] = None,
+    access_token: Optional[str] = None,
+) -> Dict[str, Any]:
+    headers = build_headers(access_token)
+    params = {
+        "method": "GET",
+        "url": "/assistants",
+        "headers": headers,
+    }
+    if assistant_id:
+        params["params"] = {"assistant_id": assistant_id}
+    return params
+
+
+def _build_response(response: httpx.Response) -> Assistants:
+    if response.status_code == HTTPStatus.OK:
+        return Assistants(**response.json())
+    elif response.status_code == HTTPStatus.UNAUTHORIZED:
+        raise AuthenticationError(response.url, response.status_code, response.content, response.headers)
+    else:
+        raise ResponseError(response.url, response.status_code, response.content, response.headers)
+
+
+def sync(
+    client: httpx.Client,
+    *,
+    assistant_id: Optional[str] = None,
+    access_token: Optional[str] = None,
+) -> Assistants:
+    """Возвращает массив объектов с данными доступных ассистентов"""
+    kwargs = _get_kwargs(assistant_id=assistant_id, access_token=access_token)
+    response = client.request(**kwargs)
+    return _build_response(response)
+
+
+async def asyncio(
+    client: httpx.AsyncClient,
+    *,
+    assistant_id: Optional[str] = None,
+    access_token: Optional[str] = None,
+) -> Assistants:
+    """Возвращает массив объектов с данными доступных ассистентов"""
+    kwargs = _get_kwargs(assistant_id=assistant_id, access_token=access_token)
+    response = await client.request(**kwargs)
+    return _build_response(response)

src/gigachat/api/assistants/post_assistant_files_delete.py

@@ -0,0 +1,60 @@
+from http import HTTPStatus
+from typing import Any, Dict, Optional
+
+import httpx
+
+from gigachat.api.utils import build_headers
+from gigachat.exceptions import AuthenticationError, ResponseError
+from gigachat.models.assistants import AssistantFileDelete
+
+
+def _get_kwargs(
+    *,
+    assistant_id: str,
+    file_id: str,
+    access_token: Optional[str] = None,
+) -> Dict[str, Any]:
+    headers = build_headers(access_token)
+
+    return {
+        "method": "POST",
+        "url": "/assistants/files/delete",
+        "json": {
+            "assistant_id": assistant_id,
+            "file_id": file_id,
+        },
+        "headers": headers,
+    }
+
+
+def _build_response(response: httpx.Response) -> AssistantFileDelete:
+    if response.status_code == HTTPStatus.OK:
+        return AssistantFileDelete(**response.json())
+    elif response.status_code == HTTPStatus.UNAUTHORIZED:
+        raise AuthenticationError(response.url, response.status_code, response.content, response.headers)
+    else:
+        raise ResponseError(response.url, response.status_code, response.content, response.headers)
+
+
+def sync(
+    client: httpx.Client,
+    *,
+    assistant_id: str,
+    file_id: str,
+    access_token: Optional[str] = None,
+) -> AssistantFileDelete:
+    kwargs = _get_kwargs(assistant_id=assistant_id, file_id=file_id, access_token=access_token)
+    response = client.request(**kwargs)
+    return _build_response(response)
+
+
+async def asyncio(
+    client: httpx.AsyncClient,
+    *,
+    assistant_id: str,
+    file_id: str,
+    access_token: Optional[str] = None,
+) -> AssistantFileDelete:
+    kwargs = _get_kwargs(assistant_id=assistant_id, file_id=file_id, access_token=access_token)
+    response = await client.request(**kwargs)
+    return _build_response(response)

src/gigachat/api/assistants/post_assistant_modify.py

@@ -0,0 +1,92 @@
+from http import HTTPStatus
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from gigachat.api.utils import build_headers
+from gigachat.exceptions import AuthenticationError, ResponseError
+from gigachat.models.assistants import Assistant
+
+
+def _get_kwargs(
+    *,
+    assistant_id: str,
+    name: Optional[str] = None,
+    description: Optional[str] = None,
+    instructions: Optional[str] = None,
+    file_ids: Optional[List[str]] = None,
+    metadata: Optional[Dict[str, Any]] = None,
+    access_token: Optional[str] = None,
+) -> Dict[str, Any]:
+    headers = build_headers(access_token)
+
+    return {
+        "method": "POST",
+        "url": "/assistants/modify",
+        "json": {
+            "assistant_id": assistant_id,
+            "name": name,
+            "description": description,
+            "instructions": instructions,
+            "file_ids": file_ids,
+            "metadata": metadata,
+        },
+        "headers": headers,
+    }
+
+
+def _build_response(response: httpx.Response) -> Assistant:
+    if response.status_code == HTTPStatus.OK:
+        return Assistant(**response.json())
+    elif response.status_code == HTTPStatus.UNAUTHORIZED:
+        raise AuthenticationError(response.url, response.status_code, response.content, response.headers)
+    else:
+        raise ResponseError(response.url, response.status_code, response.content, response.headers)
+
+
+def sync(
+    client: httpx.Client,
+    *,
+    assistant_id: str,
+    name: Optional[str] = None,
+    description: Optional[str] = None,
+    instructions: Optional[str] = None,
+    file_ids: Optional[List[str]] = None,
+    metadata: Optional[Dict[str, Any]] = None,
+    access_token: Optional[str] = None,
+) -> Assistant:
+    kwargs = _get_kwargs(
+        assistant_id=assistant_id,
+        name=name,
+        description=description,
+        instructions=instructions,
+        file_ids=file_ids,
+        metadata=metadata,
+        access_token=access_token,
+    )
+    response = client.request(**kwargs)
+    return _build_response(response)
+
+
+async def asyncio(
+    client: httpx.AsyncClient,
+    *,
+    assistant_id: str,
+    name: Optional[str] = None,
+    description: Optional[str] = None,
+    instructions: Optional[str] = None,
+    file_ids: Optional[List[str]] = None,
+    metadata: Optional[Dict[str, Any]] = None,
+    access_token: Optional[str] = None,
+) -> Assistant:
+    kwargs = _get_kwargs(
+        assistant_id=assistant_id,
+        name=name,
+        description=description,
+        instructions=instructions,
+        file_ids=file_ids,
+        metadata=metadata,
+        access_token=access_token,
+    )
+    response = await client.request(**kwargs)
+    return _build_response(response)

src/gigachat/api/assistants/post_assistants.py

@@ -0,0 +1,94 @@
+from http import HTTPStatus
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from gigachat.api.utils import build_headers
+from gigachat.exceptions import AuthenticationError, ResponseError
+from gigachat.models.assistants import CreateAssistant
+
+
+def _get_kwargs(
+    *,
+    model: str,
+    name: str,
+    description: Optional[str] = None,
+    instructions: str,
+    file_ids: Optional[List[str]] = None,
+    metadata: Optional[Dict[str, Any]] = None,
+    access_token: Optional[str] = None,
+) -> Dict[str, Any]:
+    headers = build_headers(access_token)
+
+    return {
+        "method": "POST",
+        "url": "/assistants",
+        "json": {
+            "model": model,
+            "name": name,
+            "description": description,
+            "instructions": instructions,
+            "file_ids": file_ids,
+            "metadata": metadata,
+        },
+        "headers": headers,
+    }
+
+
+def _build_response(response: httpx.Response) -> CreateAssistant:
+    if response.status_code == HTTPStatus.OK:
+        return CreateAssistant(**response.json())
+    elif response.status_code == HTTPStatus.UNAUTHORIZED:
+        raise AuthenticationError(response.url, response.status_code, response.content, response.headers)
+    else:
+        raise ResponseError(response.url, response.status_code, response.content, response.headers)
+
+
+def sync(
+    client: httpx.Client,
+    *,
+    model: str,
+    name: str,
+    description: Optional[str] = None,
+    instructions: str,
+    file_ids: Optional[List[str]] = None,
+    metadata: Optional[Dict[str, Any]] = None,
+    access_token: Optional[str] = None,
+) -> CreateAssistant:
+    """Создание ассистента"""
+    kwargs = _get_kwargs(
+        model=model,
+        name=name,
+        description=description,
+        instructions=instructions,
+        file_ids=file_ids,
+        metadata=metadata,
+        access_token=access_token,
+    )
+    response = client.request(**kwargs)
+    return _build_response(response)
+
+
+async def asyncio(
+    client: httpx.AsyncClient,
+    *,
+    model: str,
+    name: str,
+    description: Optional[str] = None,
+    instructions: str,
+    file_ids: Optional[List[str]] = None,
+    metadata: Optional[Dict[str, Any]] = None,
+    access_token: Optional[str] = None,
+) -> CreateAssistant:
+    """Создание ассистента"""
+    kwargs = _get_kwargs(
+        model=model,
+        name=name,
+        description=description,
+        instructions=instructions,
+        file_ids=file_ids,
+        metadata=metadata,
+        access_token=access_token,
+    )
+    response = await client.request(**kwargs)
+    return _build_response(response)

src/gigachat/api/post_files.py

@@ -0,0 +1,59 @@
+from http import HTTPStatus
+from typing import Any, Dict, Literal, Optional
+
+import httpx
+
+from gigachat._types import FileTypes
+from gigachat.api.utils import build_headers
+from gigachat.exceptions import AuthenticationError, ResponseError
+from gigachat.models import UploadedFile
+
+
+def _get_kwargs(
+    *,
+    file: FileTypes,
+    purpose: Literal["general", "assistant"] = "general",
+    access_token: Optional[str] = None,
+) -> Dict[str, Any]:
+    headers = build_headers(access_token)
+
+    return {
+        "method": "POST",
+        "url": "/files",
+        "files": {"file": file},
+        "data": {"purpose": purpose},
+        "headers": headers,
+    }
+
+
+def _build_response(response: httpx.Response) -> UploadedFile:
+    if response.status_code == HTTPStatus.OK:
+        return UploadedFile(**response.json())
+    elif response.status_code == HTTPStatus.UNAUTHORIZED:
+        raise AuthenticationError(response.url, response.status_code, response.content, response.headers)
+    else:
+        raise ResponseError(response.url, response.status_code, response.content, response.headers)
+
+
+def sync(
+    client: httpx.Client,
+    *,
+    file: FileTypes,
+    purpose: Literal["general", "assistant"] = "general",
+    access_token: Optional[str] = None,
+) -> UploadedFile:
+    kwargs = _get_kwargs(file=file, purpose=purpose, access_token=access_token)
+    response = client.request(**kwargs)
+    return _build_response(response)
+
+
+async def asyncio(
+    client: httpx.AsyncClient,
+    *,
+    file: FileTypes,
+    purpose: Literal["general", "assistant"] = "general",
+    access_token: Optional[str] = None,
+) -> UploadedFile:
+    kwargs = _get_kwargs(file=file, purpose=purpose, access_token=access_token)
+    response = await client.request(**kwargs)
+    return _build_response(response)