Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,10 @@ All notable changes to this project will be documented in this file.

This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).

## [v3.7.1], 2026-07-02
### Added
- Add a query_format parameter to sync/async get_attachments methods (#120 thanks @nazywam).

## [v3.7.0], 2026-06-29
### Removed
- Remove support for now EoL Python 3.9.
Expand Down
24 changes: 22 additions & 2 deletions rt/rest2.py
Original file line number Diff line number Diff line change
Expand Up @@ -859,6 +859,7 @@ def get_attachments(
self,
ticket_id: str | int,
query_filter: list[dict[str, str]] | None = None,
query_format: str | list[str] | dict[str, str] | None = None,
) -> typing.Sequence[dict[str, str]]:
"""Get attachment list for a given ticket.

Expand All @@ -879,6 +880,7 @@ def get_attachments(

:param ticket_id: ID of ticket
:param query_filter: JSON search filter, defaults to "filename is not empty"
:param query_format: Returned fields to be populated
:returns: List of tuples for attachments belonging to given ticket.
Tuple format: (id, name, content_type, size)
Returns None if ticket does not exist.
Expand All @@ -888,10 +890,18 @@ def get_attachments(
if query_filter is None:
query_filter = [{'field': 'Filename', 'operator': 'IS NOT', 'value': ''}]

get_params = {'fields': 'Filename,ContentType,ContentLength'}
if isinstance(query_format, dict):
get_params = {**get_params, **query_format}
elif isinstance(query_format, list):
get_params['fields'] = ','.join(query_format)
elif isinstance(query_format, str):
get_params['fields'] = query_format

for item in self.__paged_request(
f'ticket/{ticket_id}/attachments',
json_data=query_filter,
params={'fields': 'Filename,ContentType,ContentLength'},
params=get_params,
):
attachments.append(item)

Expand Down Expand Up @@ -2561,6 +2571,7 @@ async def get_attachments(
self,
ticket_id: str | int,
query_filter: list[dict[str, str]] | None = None,
query_format: str | list[str] | dict[str, str] | None = None,
) -> collections.abc.AsyncIterator[dict[str, typing.Any]]:
"""Get attachment list for a given ticket.

Expand All @@ -2581,15 +2592,24 @@ async def get_attachments(

:param ticket_id: ID of ticket
:param query_filter: JSON search filter, defaults to "filename is not empty"
:param query_format: Returned fields to be populated
:returns: Iterator of attachments belonging to given ticket. collections.abc.AsyncIterator[typing.Dict[str, str]]
"""
if query_filter is None:
query_filter = [{'field': 'Filename', 'operator': 'IS NOT', 'value': ''}]

get_params = {'fields': 'Filename,ContentType,ContentLength'}
if isinstance(query_format, dict):
get_params = {**get_params, **query_format}
elif isinstance(query_format, list):
get_params['fields'] = ','.join(query_format)
elif isinstance(query_format, str):
get_params['fields'] = query_format

async for item in self.__paged_request(
f'ticket/{ticket_id}/attachments',
json_data=query_filter,
params={'fields': 'Filename,ContentType,ContentLength'},
params=get_params,
):
yield item

Expand Down
23 changes: 22 additions & 1 deletion tests/test_basic.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
"""
__docformat__ = 'reStructuredText en'
__authors__ = [
'"Jiri Machalek" <jiri.machalek@nirt_connection.cz>',
'"Jiri Machalek" <jiri.machalek@nic.cz>',
'"Georges Toth" <[email protected]>',
]

Expand Down Expand Up @@ -251,6 +251,27 @@ def test_attachments_create(rt_connection: rt.rest2.Rt):
assert at_content == k.file_content


def test_attachments_search(rt_connection: rt.rest2.Rt):
"""Create a ticket with an attachments and verify that attachment search and filtering works correctly."""
ticket_subject = f'Testing issue {random_string()}'
ticket_text = (
'Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.'
)

attachment_content = random_string(length=100).encode()
attachment_name = f'attachment-{random_string(length=10)}.txt'
attachment = rt.rest2.Attachment(attachment_name, 'text/plain', attachment_content)

ticket_id = rt_connection.create_ticket(subject=ticket_subject, content=ticket_text, queue=RT_QUEUE, attachments=[attachment])

at_list = [item for item in rt_connection.get_attachments(ticket_id, query_format=['TransactionId', 'Headers'])]
assert at_list
assert len(at_list) == 1

assert at_list[0]['TransactionId']
assert at_list[0]['Headers']


def test_attachments_comment(rt_connection: rt.rest2.Rt):
"""Create a ticket and comment to it with a random (>= 2) number of attachments and verify that they have been successfully added to the ticket."""
ticket_subject = f'Testing issue {random_string()}'
Expand Down
26 changes: 25 additions & 1 deletion tests/test_basic_async.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
"""
__docformat__ = 'reStructuredText en'
__authors__ = [
'"Jiri Machalek" <jiri.machalek@niasync_rt_connection.cz>',
'"Jiri Machalek" <jiri.machalek@nic.cz>',
'"Georges Toth" <[email protected]>',
]

Expand Down Expand Up @@ -259,6 +259,30 @@ async def test_attachments_create(async_rt_connection: rt.rest2.AsyncRt):
assert at_content == k.file_content


@pytest.mark.asyncio
async def test_attachments_search(async_rt_connection: rt.rest2.AsyncRt):
"""Create a ticket with an attachments and verify that attachment search and filtering works correctly."""
ticket_subject = f'Testing issue {random_string()}'
ticket_text = (
'Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.'
)

attachment_content = random_string(length=100).encode()
attachment_name = f'attachment-{random_string(length=10)}.txt'
attachment = rt.rest2.Attachment(attachment_name, 'text/plain', attachment_content)

ticket_id = await async_rt_connection.create_ticket(
subject=ticket_subject, content=ticket_text, queue=RT_QUEUE, attachments=[attachment]
)

at_list = [item async for item in async_rt_connection.get_attachments(ticket_id, query_format=['TransactionId', 'Headers'])]
assert at_list
assert len(at_list) == 1

assert at_list[0]['TransactionId']
assert at_list[0]['Headers']


@pytest.mark.asyncio
async def test_attachments_comment(async_rt_connection: rt.rest2.AsyncRt):
"""Create a ticket and comment to it with a random (>= 2) number of attachments and verify that they have been successfully added to the ticket."""
Expand Down