Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add count param in methods #94

Merged
merged 5 commits into from
Jul 21, 2023
Merged

Add count param in methods #94

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5181eef
add count param in methods
Repumba Apr 11, 2023
386218d
fix older_than bug
Repumba Apr 11, 2023
7887ec6
ensure that count param is a string (in request)
Repumba Apr 11, 2023
91ea45f
I have misunderstood issue - now it's ok
Repumba Apr 11, 2023
cc3e176
Apply suggestions from code review
psrok1 Jul 21, 2023
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
63 changes: 46 additions & 17 deletions mwdblib/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,10 @@ def logout(self) -> None:
self.api.logout()

def _recent(
self, object_type: Type[MWDBObjectVar], query: Optional[str] = None
self,
object_type: Type[MWDBObjectVar],
query: Optional[str] = None,
chunk_size: Optional[int] = None,
) -> Iterator[MWDBObjectVar]:
"""
Generic implementation of recent_* methods
Expand All @@ -161,6 +164,8 @@ def _recent(
params = {"older_than": last_object.id} if last_object else {}
if query is not None:
params["query"] = query
if chunk_size is not None:
params["count"] = str(chunk_size)
# 'object', 'file', 'config' or 'blob'?
result = self.api.get(object_type.URL_TYPE, params=params)
key = object_type.URL_TYPE + "s"
Expand All @@ -172,7 +177,7 @@ def _recent(
except ObjectNotFoundError:
return

def recent_objects(self) -> Iterator[MWDBObject]:
def recent_objects(self, chunk_size: Optional[int] = None) -> Iterator[MWDBObject]:
"""
Retrieves recently uploaded objects
If you already know type of object you are looking for,
Expand All @@ -196,37 +201,45 @@ def recent_objects(self) -> Iterator[MWDBObject]:
files = islice(mwdb.recent_files(), 25)
print([(f.name, f.tags) for f in files])

:param chunk_size: Number of objects returned per API request
:type chunk_size: int
:rtype: Iterator[:class:`MWDBObject`]
:raises: requests.exceptions.HTTPError
"""
return self._recent(MWDBObject)
return self._recent(MWDBObject, chunk_size=chunk_size)

def recent_files(self) -> Iterator[MWDBFile]:
def recent_files(self, chunk_size: Optional[int] = None) -> Iterator[MWDBFile]:
"""
Retrieves recently uploaded files

:param chunk_size: Number of files returned per API request
:type chunk_size: int
:rtype: Iterator[:class:`MWDBFile`]
:raises: requests.exceptions.HTTPError
"""
return self._recent(MWDBFile)
return self._recent(MWDBFile, chunk_size=chunk_size)

def recent_configs(self) -> Iterator[MWDBConfig]:
def recent_configs(self, chunk_size: Optional[int] = None) -> Iterator[MWDBConfig]:
"""
Retrieves recently uploaded configuration objects

:param chunk_size: Number of configs returned per API request
:type chunk_size: int
:rtype: Iterator[:class:`MWDBConfig`]
:raises: requests.exceptions.HTTPError
"""
return self._recent(MWDBConfig)
return self._recent(MWDBConfig, chunk_size=chunk_size)

def recent_blobs(self) -> Iterator[MWDBBlob]:
def recent_blobs(self, chunk_size: Optional[int] = None) -> Iterator[MWDBBlob]:
"""
Retrieves recently uploaded blob objects

:param chunk_size: Number of blobs returned per API request
:type chunk_size: int
:rtype: Iterator[:class:`MWDBBlob`]
:raises: requests.exceptions.HTTPError
"""
return self._recent(MWDBBlob)
return self._recent(MWDBBlob, chunk_size=chunk_size)

def _listen(
self,
Expand Down Expand Up @@ -574,7 +587,9 @@ def query_blob(self, hash: str, raise_not_found: bool = True) -> Optional[MWDBBl
"""
return self._query(MWDBBlob, hash, raise_not_found)

def search(self, query: str) -> Iterator[MWDBObject]:
def search(
self, query: str, chunk_size: Optional[int] = None
) -> Iterator[MWDBObject]:
"""
Advanced search for objects using Lucene syntax.
If you already know type of objects you are looking for,
Expand All @@ -595,43 +610,57 @@ def search(self, query: str) -> Iterator[MWDBObject]:

:param query: Search query
:type query: str
:param chunk_size: Number of objects returned per API request
:type chunk_size: int
:rtype: Iterator[:class:`MWDBObject`]
:raises: requests.exceptions.HTTPError
"""
return self._recent(MWDBObject, query)
return self._recent(MWDBObject, query, chunk_size=chunk_size)

def search_files(self, query: str) -> Iterator[MWDBFile]:
def search_files(
self, query: str, chunk_size: Optional[int] = None
) -> Iterator[MWDBFile]:
"""
Advanced search for files using Lucene syntax.

:param query: Search query
:type query: str
:param chunk_size: Number of files returned per API request
:type chunk_size: int
:rtype: Iterator[:class:`MWDBFile`]
:raises: requests.exceptions.HTTPError
"""
return self._recent(MWDBFile, query)
return self._recent(MWDBFile, query, chunk_size=chunk_size)

def search_configs(self, query: str) -> Iterator[MWDBConfig]:
def search_configs(
self, query: str, chunk_size: Optional[int] = None
) -> Iterator[MWDBConfig]:
"""
Advanced search for configuration objects using Lucene syntax.

:param query: Search query
:type query: str
:param chunk_size: Number of configs returned per API request
:type chunk_size: int
:rtype: Iterator[:class:`MWDBConfig`]
:raises: requests.exceptions.HTTPError
"""
return self._recent(MWDBConfig, query)
return self._recent(MWDBConfig, query, chunk_size=chunk_size)

def search_blobs(self, query: str) -> Iterator[MWDBBlob]:
def search_blobs(
self, query: str, chunk_size: Optional[int] = None
) -> Iterator[MWDBBlob]:
"""
Advanced search for blob objects using Lucene syntax.

:param query: Search query
:type query: str
:param chunk_size: Number of blobs returned per API request
:type chunk_size: int
:rtype: Iterator[:class:`MWDBBlob`]
:raises: requests.exceptions.HTTPError
"""
return self._recent(MWDBBlob, query)
return self._recent(MWDBBlob, query, chunk_size=chunk_size)

def _count(
self, object_type: Type[MWDBObjectVar], query: Optional[str] = None
Expand Down