Merge

examples/README.md

@@ -9,3 +9,5 @@
 * [Пример вопрос - ответ, для проверки сертификатов используем файл с корневым сертификатом Минцифры России](./example_russian_trusted_root_ca.py)
 * [Пример перенаправления заголовков запроса в GigaChat](./example_contextvars.py)
 * [Пример подсчёта токенов](./example_tokens.py)
+* [Пример получения изображения](./example_get_image.py)
+* [Jupyter-ноутбук с примером работы с чатом и изображениями](./simple_chat_with_images.ipynb)

examples/example_get_image.py

@@ -0,0 +1,26 @@
+import base64
+
+from gigachat import GigaChat
+from gigachat.models import Image
+
+
+def get_image(file_path: str = './giga_img.jpg'):
+    with GigaChat(credentials=...,
+                  verify_ssl_certs=False) as giga:
+        response: Image = giga.get_image(file_id=...)
+
+        # Сохранить изображение в файл
+        with open(file_path, mode="wb") as fd:
+            fd.write(base64.b64decode(response.content))
+
+
+async def get_image_async(file_path: str = './giga_img.jpg'):
+    from aiofile import async_open
+
+    async with GigaChat(credentials=...,
+                        verify_ssl_certs=False) as giga:
+        response: Image = await giga.aget_image(file_id=...)
+
+        # Сохранить изображение в файл
+        async with async_open(file_path, 'wb') as afp:
+            await afp.write(base64.b64decode(response.content))

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "gigachat"
-version = "0.1.22"
+version = "0.1.29"
 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)