Yandex Wiki API usage examples

import os
import requests
import sys

WIKI_PUBLIC_API = 'https://api.wiki.yandex.net/v1'
WIKI_TOKEN = os.environ.get('WIKI_TOKEN')
ORG_ID = os.environ.get('ORG_ID')

HEADERS = {
    'Authorization': f'OAuth {WIKI_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def create_wiki_page(
    slug: str,
    title: str,
    content: str,
) -> int:
    """
    Create a new page

    Parameters:
    - slug: slug of the page being created
    - title: page title
    - content: page text
    """
    url = f'{WIKI_PUBLIC_API}/pages'
    body = {
        'slug': slug,
        'title': title,
        'content': content,
    }
    response = requests.post(url, json=body, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to create page: {response.status_code} {response.text}')
        return None

    res = response.json()
    return res.get('id')


def change_wiki_page(
    page_id: int,
    title: str = None,
    content: str = None,
) -> bool:
    """
    Edit an existing page

    Parameters:
    - page_id: page code
    - title: page title
    - content: page text
    """
    url = f'{WIKI_PUBLIC_API}/pages/{page_id}'
    body = {}
    if title:
        body['title'] = title
    if content:
        body['content'] = content
    response = requests.post(url, json=body, headers=HEADERS)
    return response.status_code == 200


def append_wiki_page_content(
    page_id: int,
    content: str,
    location: str = None,
) -> bool:
    """
    Add content to an existing page

    Parameters:
    - page_id: page code
    - content: text to insert
    - location: position to insert text;
      you can insert at the top of the page using `top`, at the bottom using `bottom`, or under an anchor, for example `#fragment`
    """
    url = f'{WIKI_PUBLIC_API}/pages/{page_id}/append-content'
    body = {
        'content': content,
    }
    if location in ('top', 'bottom'):
        body['body'] = {'location': location}
    elif isinstance(location, str) and location.startswith('#'):
        body['anchor'] = {'name': location}
    else:
        body['body'] = {'location': 'bottom'}
    response = requests.post(url, json=body, headers=HEADERS)
    return response.status_code == 200


def get_wiki_page_attributes(slug: str) -> int:
    """
    Get the attributes of an existing page by its slug

    Parameters:
    - slug: slug of an existing page
    """
    url = f'{WIKI_PUBLIC_API}/pages'
    params = {
        'slug': slug,
    }
    response = requests.get(url, params=params, headers=HEADERS)
    if response.status_code == 200:
        return response.json()


def main():
    if len(sys.argv) < 2:
        return
    slug = sys.argv[1]

    # Create a new page
    new_page_id = create_wiki_page(
        slug=slug,
        title='Test page',
        content=(
            'Lorem ipsum dolor sit amet, consectetur adipiscing elit, '
            'sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.\n'
        ),
    )
    if new_page_id:
        print(f'Created page with code {new_page_id}')

    # Update page content (note: this method replaces the entire page)
    is_changed = change_wiki_page(
        page_id=new_page_id,
        title='Updated test page',
        content=(
            'Lorem ipsum dolor sit amet, consectetur adipiscing elit, '
            'sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.\n'
            'Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris '
            'nisi ut aliquip ex ea commodo consequat.\n'
            'Duis aute irure dolor in reprehenderit in voluptate velit esse cillum '
            'dolore eu fugiat nulla pariatur.\n'
            'Excepteur sint occaecat cupidatat non proident, '
            'sunt in culpa qui officia deserunt mollit anim id est laborum.\n'
        ),
    )
    if is_changed:
        print('Page content updated')

    # Retrieve page attributes by its slug
    page_attributes = get_wiki_page_attributes(slug=slug)
    if not page_attributes:
        print('Wiki page not found')
        sys.exit(1)
    page_id = page_attributes.get('id')
    print(f'Found page with code {page_id}')


if __name__ == '__main__':
    main()

import os
import requests
import sys

WIKI_PUBLIC_API = 'https://api.wiki.yandex.net/v1'
WIKI_TOKEN = os.environ.get('WIKI_TOKEN')
ORG_ID = os.environ.get('ORG_ID')

HEADERS = {
    'Authorization': f'OAuth {WIKI_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def get_wiki_page_attributes(slug: str) -> int:
    """
    Get the attributes of an existing page by its slug

    Parameters:
    - slug: slug of an existing page
    """
    url = f'{WIKI_PUBLIC_API}/pages'
    params = {
        'slug': slug,
    }
    response = requests.get(url, params=params, headers=HEADERS)
    if response.status_code == 200:
        return response.json()


def append_wiki_page_content(
    page_id: int,
    content: str,
    location: str = None,
) -> bool:
    """
    Add content to an existing page

    Parameters:
    - page_id: page code
    - content: text to insert
    - location: position to insert text;
      you can insert at the top of the page using `top`, at the bottom using `bottom`,
      or under an anchor, for example `#fragment`
    """
    url = f'{WIKI_PUBLIC_API}/pages/{page_id}/append-content'
    body = {
        'content': content,
    }
    if location in ('top', 'bottom'):
        body['body'] = {'location': location}
    elif isinstance(location, str) and location.startswith('#'):
        body['anchor'] = {'name': location}
    else:
        body['body'] = {'location': 'bottom'}
    response = requests.post(url, json=body, headers=HEADERS)
    return response.status_code == 200


def create_wiki_grid(page_id: int, title: str) -> int:
    """
    Create a new table

    Parameters:
    - page_id: page code
    - title: table title

    Note: after creation, the table is only accessible through the resource view interface;
    you still need to add it to a page.
    """
    url = f'{WIKI_PUBLIC_API}/grids'
    body = {
        'page': {'id': page_id},
        'title': title,
    }
    response = requests.post(url, json=body, headers=HEADERS)
    if response.status_code == 200:
        res = response.json()
        return res.get('id')


def insert_grid_columns(grid_id: str, columns: list[dict]) -> bool:
    """
    Add columns to a table

    Parameters:
    - grid_id: table code
    - columns: list of columns to add
    """
    url = f'{WIKI_PUBLIC_API}/grids/{grid_id}/columns'
    body = {
        'columns': columns,
    }
    response = requests.post(url, json=body, headers=HEADERS)
    return response.status_code == 200


def insert_grid_rows(grid_id: str, rows: list[dict]) -> bool:
    """
    Add rows to a table

    Parameters:
    - grid_id: table code
    - rows: list of rows to add
    """
    url = f'{WIKI_PUBLIC_API}/grids/{grid_id}/rows'
    body = {
        'rows': rows,
    }
    response = requests.post(url, json=body, headers=HEADERS)
    return response.status_code == 200


def change_grid_cells(grid_id: str, cells: list[dict]) -> bool:
    """
    Edit cells in a table

    Parameters:
    - grid_id: table code
    - cells: list of cells to edit
    """
    url = f'{WIKI_PUBLIC_API}/grids/{grid_id}/cells'
    body = {
        'cells': cells,
    }
    response = requests.post(url, json=body, headers=HEADERS)
    return response.status_code == 200


def main():
    if len(sys.argv) < 2:
        return
    slug = sys.argv[1]

    # Retrieve page attributes by its slug
    page_attributes = get_wiki_page_attributes(
        slug=slug,
    )
    if not page_attributes:
        print('Wiki page not found')
        sys.exit(1)
    page_id = page_attributes.get('id')
    print(f'Found page with code {page_id}')

    # Create a new table
    grid_id = create_wiki_grid(
        page_id=page_id,
        title='New table',
    )
    if grid_id:
        print(f'Created table with code {grid_id}')

    # Add columns to the new table (columns can be of different types)
    is_changed = insert_grid_columns(
        grid_id=grid_id,
        columns=[
            {
                'slug': 'id',
                'title': 'ID',
                'type': 'number',
                'required': True,
            },
            {
                'slug': 'name',
                'title': 'Name',
                'type': 'string',
                'required': False,
            },
        ],
    )
    if is_changed:
        print('New columns added to table')

    # Add rows to the new table
    is_changed = insert_grid_rows(
        grid_id=grid_id,
        rows=[
            {'id': 1, 'name': 'One'},
            {'id': 2, 'name': 'Two'},
            {'id': 3, 'name': 'Three'},
            {'id': 4, 'name': 'Four'},
            {'id': 5, 'name': 'Five'},
        ],
    )
    if is_changed:
        print('New rows added to table')

    # Edit cells in the second row of the table
    is_changed = change_grid_cells(
        grid_id=grid_id,
        cells=[
            {'row_id': 2, 'column_slug': 'id', 'value': 22},
            {'row_id': 2, 'column_slug': 'name', 'value': 'Twenty-two'},
        ],
    )
    if is_changed:
        print('Second row in table updated')

    # Place the new table at the bottom of the page
    is_changed = append_wiki_page_content(
        page_id=page_id,
        content=f'\n{{% wgrid id="{grid_id}" %}}',
    )
    if is_changed:
        print('Created table added to the bottom of the page')


if __name__ == '__main__':
    main()

import os
import requests
import sys

WIKI_PUBLIC_API = 'https://api.wiki.yandex.net/v1'
WIKI_TOKEN = os.environ.get('WIKI_TOKEN')
ORG_ID = os.environ.get('ORG_ID')

HEADERS = {
    'Authorization': f'OAuth {WIKI_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def create_wiki_page(slug: str, title: str, content: str) -> int | None:
    """
    Create a new page
    """
    url = f'{WIKI_PUBLIC_API}/pages'
    body = {
        'slug': slug,
        'title': title,
        'content': content,
    }
    response = requests.post(url, json=body, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to create page: {response.status_code} {response.text}')
        return None
    return response.json().get('id')


def get_wiki_page_id(slug: str) -> int | None:
    """
    Get the code of an existing page by its slug (address)
    """
    url = f'{WIKI_PUBLIC_API}/pages'
    response = requests.get(url, params={'slug': slug}, headers=HEADERS)
    if response.status_code != 200:
        return None
    return response.json().get('id')


def create_comment(page_id: int, body: str, parent_id: int = None, thread_id: int = None) -> int | None:
    """
    Create a comment on a page
    """
    url = f'{WIKI_PUBLIC_API}/pages/{page_id}/comments'
    data = {'body': body}
    if parent_id is not None:
        data['parent_id'] = parent_id
    if thread_id is not None:
        data['thread_id'] = thread_id
    response = requests.post(url, json=data, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to create comment: {response.status_code} {response.text}')
        return None
    return response.json().get('id')


def get_page_comments(page_id: int) -> list | None:
    """
    Get the list of comments on a page
    """
    url = f'{WIKI_PUBLIC_API}/pages/{page_id}/comments'
    response = requests.get(url, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to retrieve comments: {response.status_code} {response.text}')
        return None
    return response.json().get('results')


def main():
    if len(sys.argv) < 2:
        return
    slug = sys.argv[1]

    # Retrieve an existing page or create a new one
    page_id = get_wiki_page_id(slug=slug)
    if page_id:
        print(f'Found page with code {page_id}')
    else:
        page_id = create_wiki_page(
            slug=slug,
            title='Test page with comments',
            content='See comments.\n',
        )
        if not page_id:
            return
        print(f'Created page with code {page_id}')

    # Create two root comments and three replies:
    comment_1 = create_comment(page_id=page_id, body='First comment')
    if comment_1:
        print(f'Created comment {comment_1}')

    comment_2 = create_comment(page_id=page_id, body='Second comment')
    if comment_2:
        print(f'Created comment {comment_2}')

    comment_3 = create_comment(page_id=page_id, body='Third comment', parent_id=comment_1, thread_id=comment_1)
    if comment_3:
        print(f'Created comment {comment_3}')

    comment_4 = create_comment(page_id=page_id, body='Fourth comment', parent_id=comment_1, thread_id=comment_1)
    if comment_4:
        print(f'Created comment {comment_4}')

    comment_5 = create_comment(page_id=page_id, body='Fifth comment', parent_id=comment_2, thread_id=comment_2)
    if comment_5:
        print(f'Created comment {comment_5}')


if __name__ == '__main__':
    main()

import os
import requests
import sys

WIKI_PUBLIC_API = 'https://api.wiki.yandex.net/v1'
WIKI_TOKEN = os.environ.get('WIKI_TOKEN')
ORG_ID = os.environ.get('ORG_ID')

HEADERS = {
    'Authorization': f'OAuth {WIKI_TOKEN}',
    'X-Org-Id': ORG_ID,
}



def create_wiki_page(slug: str, title: str, content: str) -> int | None:
    """
    Create a new page
    """
    url = f'{WIKI_PUBLIC_API}/pages'
    body = {
        'slug': slug,
        'title': title,
        'content': content,
    }
    response = requests.post(url, json=body, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to create page {slug}: {response.status_code} {response.text}')
        return None
    return response.json().get('id')


def get_wiki_page_id(slug: str) -> int | None:
    """
    Get the code of an existing page by its slug
    """
    url = f'{WIKI_PUBLIC_API}/pages'
    response = requests.get(url, params={'slug': slug}, headers=HEADERS)
    if response.status_code != 200:
        return None
    return response.json().get('id')


def get_descendants(slug: str, include_self: bool = False) -> list | None:
    """
    Get the list of descendants of a page by its slug
    """
    url = f'{WIKI_PUBLIC_API}/pages/descendants'
    params = {
        'slug': slug,
        'include_self': include_self,
        'page_size': 100,
    }
    response = requests.get(url, params=params, headers=HEADERS)
    if response.status_code != 200:
        return None
    return response.json().get('results')


def main():
    if len(sys.argv) < 2:
        return
    slug = sys.argv[1]

    pages = [
        (slug, 'Test page for subpages'),
        (f'{slug}/page-1', 'Page 1'),
        (f'{slug}/page-1/sub-1', 'Subpage 1-1'),
        (f'{slug}/page-1/sub-2', 'Subpage 1-2'),
        (f'{slug}/page-2', 'Page 2'),
        (f'{slug}/page-2/sub-1', 'Subpage 2-1'),
        (f'{slug}/page-2/sub-2', 'Subpage 2-2'),
        (f'{slug}/page-3', 'Page 3'),
        (f'{slug}/page-4', 'Page 4'),
        (f'{slug}/page-5', 'Page 5'),
    ]

    # Create a page tree
    for page_slug, title in pages:
        page_id = get_wiki_page_id(slug=page_slug)
        if page_id:
            print(f'Page found: {page_slug} (code {page_id})')
        else:
            page_id = create_wiki_page(slug=page_slug, title=title, content=f'{title}.\n')
            if not page_id:
                return
            print(f'Page created: {page_slug} (code {page_id})')

    # Retrieve descendants of the specified page (including the page itself)
    print(f'Descendants of {slug} (inclusive):')
    subtree = get_descendants(slug=slug, include_self=True)
    if subtree is not None:
        for page in subtree:
            print(f'- {page.get("slug")}')
        print(f'Total: {len(subtree)}')


if __name__ == '__main__':
    main()

import os
import requests
import sys

WIKI_PUBLIC_API = 'https://api.wiki.yandex.net/v1'
WIKI_TOKEN = os.environ.get('WIKI_TOKEN')
ORG_ID = os.environ.get('ORG_ID')

HEADERS = {
    'Authorization': f'OAuth {WIKI_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def create_wiki_page(slug: str, title: str, content: str) -> int | None:
    """
    Create a new page
    """
    url = f'{WIKI_PUBLIC_API}/pages'
    body = {
        'slug': slug,
        'title': title,
        'content': content,
    }
    response = requests.post(url, json=body, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to create page: {response.status_code} {response.text}')
        return None
    return response.json().get('id')


def get_wiki_page_id(slug: str) -> int | None:
    """
    Get the code of an existing page by its slug
    """
    url = f'{WIKI_PUBLIC_API}/pages'
    response = requests.get(url, params={'slug': slug}, headers=HEADERS)
    if response.status_code != 200:
        return None
    return response.json().get('id')


def delete_wiki_page(page_id: int) -> str | None:
    """
    Delete a page and return the recovery token
    """
    url = f'{WIKI_PUBLIC_API}/pages/{page_id}'
    response = requests.delete(url, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to delete page: {response.status_code} {response.text}')
        return None
    return response.json().get('recovery_token')


def recover_wiki_page(token: str) -> int | None:
    """
    Recover a deleted page using a recovery token
    """
    url = f'{WIKI_PUBLIC_API}/recovery_tokens/{token}/recover'
    response = requests.post(url, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to recover page: {response.status_code} {response.text}')
        return None
    return response.json().get('id')


def main():
    if len(sys.argv) < 2:
        return
    slug = sys.argv[1]

    # Retrieve an existing page or create a new one
    page_id = get_wiki_page_id(slug=slug)
    if page_id:
        print(f'Found page with code {page_id}')
    else:
        page_id = create_wiki_page(
            slug=slug,
            title='Test page for recovery',
            content='Test page content.\n',
        )
        if not page_id:
            return
        print(f'Created page with code {page_id}')

    # Delete the page and obtain the recovery token
    recovery_token = delete_wiki_page(page_id=page_id)
    if not recovery_token:
        return
    print(f'Page deleted, recovery token: {recovery_token}')

    # Recover the page using the token
    restored_id = recover_wiki_page(token=recovery_token)
    if not restored_id:
        return
    print(f'Page recovered with code {restored_id}')


if __name__ == '__main__':
    main()

import os
import requests
import sys

WIKI_PUBLIC_API = 'https://api.wiki.yandex.net/v1'
WIKI_TOKEN = os.environ.get('WIKI_TOKEN')
ORG_ID = os.environ.get('ORG_ID')

CHUNK_SIZE = 5 * 1024 * 1024  # 5 MB — minimum part size for multipart upload

HEADERS = {
    'Authorization': f'OAuth {WIKI_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def get_file_pattern(file_url: str, file_name: str) -> str:
    return f'{{% file src="{file_url}" name="{file_name}" %}}'


def create_wiki_page(slug: str, title: str, content: str) -> int | None:
    """
    Create a new page
    """
    url = f'{WIKI_PUBLIC_API}/pages'
    body = {
        'slug': slug,
        'title': title,
        'content': content,
    }
    response = requests.post(url, json=body, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to create page: {response.status_code} {response.text}')
        return None
    return response.json().get('id')


def get_wiki_page_id(slug: str) -> int | None:
    """
    Get the code of an existing page by its slug (address)
    """
    url = f'{WIKI_PUBLIC_API}/pages'
    response = requests.get(url, params={'slug': slug}, headers=HEADERS)
    if response.status_code != 200:
        return None
    return response.json().get('id')


def create_upload_session(file_name: str, file_size: int) -> str | None:
    """
    Create a file upload session
    """
    url = f'{WIKI_PUBLIC_API}/upload_sessions'
    body = {
        'file_name': file_name,
        'file_size': file_size,
    }
    response = requests.post(url, json=body, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to create upload session: {response.status_code} {response.text}')
        return None
    return response.json().get('session_id')


def upload_file_part(session_id: str, part_number: int, data: bytes) -> bool:
    """
    Upload a part of the file within a session
    """
    url = f'{WIKI_PUBLIC_API}/upload_sessions/{session_id}/upload_part'
    headers = {**HEADERS, 'Content-Type': 'application/octet-stream'}
    response = requests.put(
        url,
        params={'part_number': part_number},
        data=data,
        headers=headers,
        stream=True,
    )
    return response.status_code == 200


def finish_upload_session(session_id: str) -> bool:
    """
    Finish the file upload session
    """
    url = f'{WIKI_PUBLIC_API}/upload_sessions/{session_id}/finish'
    response = requests.post(url, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to finish upload session: {response.status_code} {response.text}')
        return False
    return True


def attach_file_to_page(page_id: int, session_id: str) -> list | None:
    """
    Attach an uploaded file to a page
    """
    url = f'{WIKI_PUBLIC_API}/pages/{page_id}/attachments'
    body = {
        'upload_sessions': [session_id],
    }
    response = requests.post(url, json=body, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to attach file to page: {response.status_code} {response.text}')
        return None
    return response.json().get('results')


def append_wiki_page_content(page_id: int, content: str) -> bool:
    """
    Add content to the bottom of a page
    """
    url = f'{WIKI_PUBLIC_API}/pages/{page_id}/append-content'
    body = {
        'content': content,
        'body': {'location': 'bottom'},
    }
    response = requests.post(url, json=body, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to add content to page: {response.status_code} {response.text}')
        return False
    return True


def get_page_resources(page_id: int) -> list | None:
    """
    Get the list of page resources (attachments, tables)
    """
    url = f'{WIKI_PUBLIC_API}/pages/{page_id}/resources'
    response = requests.get(url, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to retrieve page resources: {response.status_code} {response.text}')
        return None
    return response.json().get('results')


def get_page_attachments(page_id: int) -> list | None:
    """
    Get the list of page attachments
    """
    url = f'{WIKI_PUBLIC_API}/pages/{page_id}/attachments'
    response = requests.get(url, headers=HEADERS)
    if response.status_code != 200:
        print(f'Failed to retrieve page attachments: {response.status_code} {response.text}')
        return None
    return response.json().get('results')


def main():
    if len(sys.argv) < 3:
        return
    slug = sys.argv[1]
    file_path = sys.argv[2]

    # Retrieve an existing page or create a new one
    page_id = get_wiki_page_id(slug=slug)
    if page_id:
        print(f'Found page with code {page_id}')
    else:
        page_id = create_wiki_page(
            slug=slug,
            title='Test page with attachments',
            content='Some text.\n',
        )
        if not page_id:
            return
        print(f'Created page with code {page_id}')

    # Read the file for upload
    file_name = os.path.basename(file_path)
    file_size = os.path.getsize(file_path)

    # Create an upload session
    session_id = create_upload_session(
        file_name=file_name,
        file_size=file_size,
    )
    if not session_id:
        return
    print(f'Created upload session {session_id}')

    # Upload the file in parts (each part is at least 5 MB, except the last one)
    with open(file_path, 'rb') as f:
        part_number = 1
        while True:
            chunk = f.read(CHUNK_SIZE)
            if not chunk:
                break
            if not upload_file_part(session_id=session_id, part_number=part_number, data=chunk):
                return
            print(f'Uploaded part {part_number}')
            part_number += 1

    # Finish the upload session
    if not finish_upload_session(session_id=session_id):
        return
    print('Upload session finished')

    # Attach the file to the page
    attached = attach_file_to_page(page_id=page_id, session_id=session_id)
    if not attached:
        return
    print('File attached to page')

    # Insert a file reference at the bottom of the page
    file_url = attached[0].get('download_url')
    file_name = attached[0].get('name')
    text = f'Attached file: {get_file_pattern(file_url, file_name)}'
    if append_wiki_page_content(page_id=page_id, content=text):
        print(f'Reference to file {file_name} added to page')

    # Get the list of page resources
    resources = get_page_resources(page_id=page_id)
    if resources is not None:
        print(f'\nPage resources ({len(resources)}):')
        for resource in resources:
            resource_type = resource.get('type')
            item = resource.get('item', {})
            print(f'- Type: {resource_type}, name: {item.get('name')}')

    # Get the list of page attachments
    attachments = get_page_attachments(page_id=page_id)
    if attachments is not None:
        print(f'\nPage attachments ({len(attachments)}):')
        for attachment in attachments:
            print(
                f'- Code: {attachment.get('id')}, '
                f'name: {attachment.get('name')}, '
                f'size: {attachment.get('size')} MB, '
                f'type: {attachment.get('mimetype')}'
            )


if __name__ == '__main__':
    main()