Module neuroio.licenses.sources.v1

None

None

View Source

from datetime import datetime

from typing import Union

from httpx import Response

from neuroio.base import IAMBase, IAMBaseAsync, IAMBaseBase

from neuroio.constants import sentinel

from neuroio.utils import request_dict_processing, request_query_processing

class LicensesBase(IAMBaseBase):

    def get_url(self, key: str = None) -> str:

        if key:

            return self.base_url + f"/v1/licenses/sources/{key}/"

        else:

            return self.base_url + "/v1/licenses/sources/"

class Impl(IAMBase, LicensesBase):

    def create(self, name: str, entry_storage_days: int = 1) -> Response:

        data = request_dict_processing(locals(), ["self"])

        with self.get_client() as client:

            return client.post(url=self.get_url(), json=data)

    def list(

        self,

        q: str = "",

        date_from: Union[datetime, object] = sentinel,

        date_to: Union[datetime, object] = sentinel,

        is_bound: Union[bool, object] = sentinel,

        limit: int = 20,

        offset: int = 0,

    ) -> Response:

        data = request_query_processing(locals(), ["self"])

        with self.get_client() as client:

            return client.get(url=self.get_url(), params=data)

    def get(self, id: str) -> Response:

        with self.get_client() as client:

            return client.get(url=self.get_url(f"{id}"))

    def update(

        self,

        id: int,

        name: str,

        is_active: bool = True,

        entry_storage_days: int = 1,

    ) -> Response:

        data = request_dict_processing(locals(), ["self", "id"])

        with self.get_client() as client:

            return client.patch(url=self.get_url(f"{id}"), json=data)

class ImplAsync(IAMBaseAsync, LicensesBase):

    async def create(self, name: str, entry_storage_days: int = 1) -> Response:

        data = request_dict_processing(locals(), ["self"])

        async with self.get_client() as client:

            return await client.post(url=self.get_url(), json=data)

    async def list(

        self,

        q: str = "",

        date_from: Union[datetime, object] = sentinel,

        date_to: Union[datetime, object] = sentinel,

        is_bound: Union[bool, object] = sentinel,

        limit: int = 20,

        offset: int = 0,

    ) -> Response:

        data = request_query_processing(locals(), ["self"])

        async with self.get_client() as client:

            return await client.get(url=self.get_url(), params=data)

    async def get(self, id: int) -> Response:

        async with self.get_client() as client:

            return await client.get(url=self.get_url(f"{id}"))

    async def update(

        self,

        id: int,

        name: str,

        is_active: bool = True,

        entry_storage_days: int = 1,

    ) -> Response:

        data = request_dict_processing(locals(), ["self", "id"])

        async with self.get_client() as client:

            return await client.patch(url=self.get_url(f"{id}"), json=data)

Variables

sentinel

Classes

Impl

class Impl(
    settings: dict
)

View Source

class Impl(IAMBase, LicensesBase):

    def create(self, name: str, entry_storage_days: int = 1) -> Response:

        data = request_dict_processing(locals(), ["self"])

        with self.get_client() as client:

            return client.post(url=self.get_url(), json=data)

    def list(

        self,

        q: str = "",

        date_from: Union[datetime, object] = sentinel,

        date_to: Union[datetime, object] = sentinel,

        is_bound: Union[bool, object] = sentinel,

        limit: int = 20,

        offset: int = 0,

    ) -> Response:

        data = request_query_processing(locals(), ["self"])

        with self.get_client() as client:

            return client.get(url=self.get_url(), params=data)

    def get(self, id: str) -> Response:

        with self.get_client() as client:

            return client.get(url=self.get_url(f"{id}"))

    def update(

        self,

        id: int,

        name: str,

        is_active: bool = True,

        entry_storage_days: int = 1,

    ) -> Response:

        data = request_dict_processing(locals(), ["self", "id"])

        with self.get_client() as client:

            return client.patch(url=self.get_url(f"{id}"), json=data)

---

Ancestors (in MRO)

• neuroio.base.IAMBase
• abc.ABC
• neuroio.base.Base
• neuroio.licenses.sources.v1.LicensesBase
• neuroio.base.IAMBaseBase

Class variables

base_url

Methods

create

def create(
    self,
    name: str,
    entry_storage_days: int = 1
) -> httpx.Response

View Source

    def create(self, name: str, entry_storage_days: int = 1) -> Response:

        data = request_dict_processing(locals(), ["self"])

        with self.get_client() as client:

            return client.post(url=self.get_url(), json=data)

get

def get(
    self,
    id: str
) -> httpx.Response

View Source

    def get(self, id: str) -> Response:

        with self.get_client() as client:

            return client.get(url=self.get_url(f"{id}"))

get_client

def get_client(
    self
) -> httpx.Client

View Source

    def get_client(self) -> Client:

        return Client(**self.settings)

get_url

def get_url(
    self,
    key: str = None
) -> str

View Source

    def get_url(self, key: str = None) -> str:

        if key:

            return self.base_url + f"/v1/licenses/sources/{key}/"

        else:

            return self.base_url + "/v1/licenses/sources/"

list

def list(
    self,
    q: str = '',
    date_from: Union[datetime.datetime, object] = <object object at 0x10c207fb0>,
    date_to: Union[datetime.datetime, object] = <object object at 0x10c207fb0>,
    is_bound: Union[bool, object] = <object object at 0x10c207fb0>,
    limit: int = 20,
    offset: int = 0
) -> httpx.Response

View Source

    def list(

        self,

        q: str = "",

        date_from: Union[datetime, object] = sentinel,

        date_to: Union[datetime, object] = sentinel,

        is_bound: Union[bool, object] = sentinel,

        limit: int = 20,

        offset: int = 0,

    ) -> Response:

        data = request_query_processing(locals(), ["self"])

        with self.get_client() as client:

            return client.get(url=self.get_url(), params=data)

update

def update(
    self,
    id: int,
    name: str,
    is_active: bool = True,
    entry_storage_days: int = 1
) -> httpx.Response

View Source

    def update(

        self,

        id: int,

        name: str,

        is_active: bool = True,

        entry_storage_days: int = 1,

    ) -> Response:

        data = request_dict_processing(locals(), ["self", "id"])

        with self.get_client() as client:

            return client.patch(url=self.get_url(f"{id}"), json=data)

ImplAsync

class ImplAsync(
    settings: dict
)

View Source

class ImplAsync(IAMBaseAsync, LicensesBase):

    async def create(self, name: str, entry_storage_days: int = 1) -> Response:

        data = request_dict_processing(locals(), ["self"])

        async with self.get_client() as client:

            return await client.post(url=self.get_url(), json=data)

    async def list(

        self,

        q: str = "",

        date_from: Union[datetime, object] = sentinel,

        date_to: Union[datetime, object] = sentinel,

        is_bound: Union[bool, object] = sentinel,

        limit: int = 20,

        offset: int = 0,

    ) -> Response:

        data = request_query_processing(locals(), ["self"])

        async with self.get_client() as client:

            return await client.get(url=self.get_url(), params=data)

    async def get(self, id: int) -> Response:

        async with self.get_client() as client:

            return await client.get(url=self.get_url(f"{id}"))

    async def update(

        self,

        id: int,

        name: str,

        is_active: bool = True,

        entry_storage_days: int = 1,

    ) -> Response:

        data = request_dict_processing(locals(), ["self", "id"])

        async with self.get_client() as client:

            return await client.patch(url=self.get_url(f"{id}"), json=data)

---

Ancestors (in MRO)

• neuroio.base.IAMBaseAsync
• abc.ABC
• neuroio.base.Base
• neuroio.licenses.sources.v1.LicensesBase
• neuroio.base.IAMBaseBase

Class variables

base_url

Methods

create

def create(
    self,
    name: str,
    entry_storage_days: int = 1
) -> httpx.Response

View Source

    async def create(self, name: str, entry_storage_days: int = 1) -> Response:

        data = request_dict_processing(locals(), ["self"])

        async with self.get_client() as client:

            return await client.post(url=self.get_url(), json=data)

get

def get(
    self,
    id: int
) -> httpx.Response

View Source

    async def get(self, id: int) -> Response:

        async with self.get_client() as client:

            return await client.get(url=self.get_url(f"{id}"))

get_client

def get_client(
    self
) -> httpx.AsyncClient

View Source

    def get_client(self) -> AsyncClient:

        return AsyncClient(**self.settings)

get_url

def get_url(
    self,
    key: str = None
) -> str

View Source

    def get_url(self, key: str = None) -> str:

        if key:

            return self.base_url + f"/v1/licenses/sources/{key}/"

        else:

            return self.base_url + "/v1/licenses/sources/"

list

def list(
    self,
    q: str = '',
    date_from: Union[datetime.datetime, object] = <object object at 0x10c207fb0>,
    date_to: Union[datetime.datetime, object] = <object object at 0x10c207fb0>,
    is_bound: Union[bool, object] = <object object at 0x10c207fb0>,
    limit: int = 20,
    offset: int = 0
) -> httpx.Response

View Source

    async def list(

        self,

        q: str = "",

        date_from: Union[datetime, object] = sentinel,

        date_to: Union[datetime, object] = sentinel,

        is_bound: Union[bool, object] = sentinel,

        limit: int = 20,

        offset: int = 0,

    ) -> Response:

        data = request_query_processing(locals(), ["self"])

        async with self.get_client() as client:

            return await client.get(url=self.get_url(), params=data)

update

def update(
    self,
    id: int,
    name: str,
    is_active: bool = True,
    entry_storage_days: int = 1
) -> httpx.Response

View Source

    async def update(

        self,

        id: int,

        name: str,

        is_active: bool = True,

        entry_storage_days: int = 1,

    ) -> Response:

        data = request_dict_processing(locals(), ["self", "id"])

        async with self.get_client() as client:

            return await client.patch(url=self.get_url(f"{id}"), json=data)

LicensesBase

class LicensesBase(
    /,
    *args,
    **kwargs
)

View Source

class LicensesBase(IAMBaseBase):

    def get_url(self, key: str = None) -> str:

        if key:

            return self.base_url + f"/v1/licenses/sources/{key}/"

        else:

            return self.base_url + "/v1/licenses/sources/"

---

Ancestors (in MRO)

• neuroio.base.IAMBaseBase

Descendants

• neuroio.licenses.sources.v1.Impl
• neuroio.licenses.sources.v1.ImplAsync

Class variables

base_url

Methods

get_url

def get_url(
    self,
    key: str = None
) -> str

View Source

    def get_url(self, key: str = None) -> str:

        if key:

            return self.base_url + f"/v1/licenses/sources/{key}/"

        else:

            return self.base_url + "/v1/licenses/sources/"