assistants

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 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 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,98 @@
+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 import Assistant
+
+
+def _get_kwargs(
+    *,
+    assistant_id: str,
+    model: Optional[str] = None,
+    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,
+            "model": model,
+            "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,
+    model: Optional[str] = None,
+    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,
+        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,
+    *,
+    assistant_id: str,
+    model: Optional[str] = None,
+    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,
+        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/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 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/threads/get_threads.py

@@ -0,0 +1,52 @@
+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 import Threads
+
+
+def _get_kwargs(
+    *,
+    access_token: Optional[str] = None,
+) -> Dict[str, Any]:
+    headers = build_headers(access_token)
+    params = {
+        "method": "GET",
+        "url": "/threads",
+        "headers": headers,
+    }
+    return params
+
+
+def _build_response(response: httpx.Response) -> Threads:
+    if response.status_code == HTTPStatus.OK:
+        return Threads(**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,
+    *,
+    access_token: Optional[str] = None,
+) -> Threads:
+    """Получение перечня тредов"""
+    kwargs = _get_kwargs(access_token=access_token)
+    response = client.request(**kwargs)
+    return _build_response(response)
+
+
+async def asyncio(
+    client: httpx.AsyncClient,
+    *,
+    access_token: Optional[str] = None,
+) -> Threads:
+    """Получение перечня тредов"""
+    kwargs = _get_kwargs(access_token=access_token)
+    response = await client.request(**kwargs)
+    return _build_response(response)