Module lunar.rec

Sub-modules

lunar.rec.channels
lunar.rec.experiments
lunar.rec.recommendations

Classes

class ChannelClient (config: Config)

Client for BAP Recommendation API (/v1/channels/).

Example

import bap

client = bap.client("channel")

Ancestors

Methods

def create_channel(self, id: str, experiment_id: str = None) ‑> Channel

Create a new channel.

Args

  • id: (str) Unique identifier of a channel
  • experiment_id: (optional) (str) Unique identifier of an experiment assigned

Returns

Channel

Example

channel = client.create_channel(id="my_channel", experiment_id="my_experiment")
async def create_channel_async(self, id: str, experiment_id: str = None) ‑> Channel

Create a new channel (async).

Args

  • id: (str) Unique identifier of a channel
  • experiment_id: (optional) (str) Unique identifier of an experiment assigned

Returns

Channel

Example

channel = await client.create_channel_async(id="my_channel", experiment_id="my_experiment")
def delete_channel(self, id: str) ‑> None

Delete a channel.

Args

  • id: (str) Unique identifier of a channel

Example

client.delete_channel(id="my_channel")
async def delete_channel_async(self, id: str) ‑> None

Delete a channel (async).

Args

  • id: (str) Unique identifier of a channel

Example

await client.delete_channel_async(id="my_channel")
def get_channel(self, id: str) ‑> Channel

Get a channel.

Args

  • id: (str) Unique identifier of the channel

Returns

Channel

Example

channel = client.get_channel(id="my_channel")
async def get_channel_async(self, id: str) ‑> Channel

Get a channel (async).

Args

  • id: (str) Unique identifier of a channel

Returns

Channel

Example

channel = await client.get_channel_async(id="my_channel")
def list_channels(self) ‑> List[Channel]

List channels.

Returns

list(Channel)

Example

channels = client.list_channels()
async def list_channels_async(self) ‑> List[Channel]

List channels (async).

Returns

list(Channel)

Example

channels = await client.list_channels_async()
def update_channel(self, id: str, experiment_id: str = None) ‑> Channel

Update a channel.

Args

  • id: (str) Unique identifier of a channel
  • experiment_id: (optional) (str) Unique identifier of an experiment assigned

Returns

Channel

Example

updated_channel = client.update_channel(id="my_channel", experiment_id="new_experiment")
async def update_channel_async(self, id: str, experiment_id: str = None) ‑> Channel

Update a channel (async).

Args

  • id: (str) Unique identifier of a channel
  • experiment_id: (optional) (str) Unique identifier of an experiment assigned

Returns

Channel

Example

updated_channel = await client.update_channel_async(id="my_channel", experiment_id="new_experiment")
def update_channel_partial(self, id: str, experiment_id: str = None) ‑> Channel

Partially update a channel.

Args

  • id: (str) Unique identifier of a channel
  • experiment_id: (optional) (str) Unique identifier of an experiment assigned

Returns

Channel

Example

updated_channel = client.update_channel_partial(id="my_channel", experiment_id="new_experiment")
async def update_channel_partial_async(self, id: str, experiment_id: str = None) ‑> Channel

Partially update a channel (async).

Args

  • id: (str) Unique identifier of a channel
  • experiment_id: (optional) (str) Unique identifier of an experiment assigned

Returns

Channel

Example

updated_channel = await client.update_channel_partial_async(id="my_channel", experiment_id="new_experiment")
class ExperimentClient (config: Config)

Client for BAP Recommendation API (/v1/experiments/).

Example

import bap

client = bap.client("experiment")

Ancestors

Methods

def create_experiment(self, id: str, buckets: List[Dict[str, ]], bucketing_seed: str = '') ‑> Experiment

Create a new experiment.

Args

  • id: (str) Unique identifier of an experiment
  • buckets: (list(dict)) Bucket list in an experiment
  • bucketing_seed: (optional) (str) Random seed for bucketing (default: "")

Returns

Experiment

Example

experiment = client.create_experiment(
    id="my_experiment", buckets=[{"name": "a", "ratio": 20}, {"name": "b", "ratio": 80}], bucketing_seed="100"
)
async def create_experiment_async(self, id: str, buckets: List[Dict[str, Any]], bucketing_seed: str = '') ‑> Experiment

Create a new experiment (async).

Args

  • id: (str) Unique identifier of an experiment
  • buckets: (list(dict)) Bucket list in an experiment
  • bucketing_seed: (optional) (str) Random seed for bucketing (default: "")

Returns

Experiment

Example

experiment = await client.create_experiment_async(
    id="my_experiment", buckets=[{"name": "a", "ratio": 20}, {"name": "b", "ratio": 80}], bucketing_seed="100"
)
def delete_experiment(self, id: str) ‑> None

Delete an experiment.

Args

  • id: (str) Unique identifier of an experiment

Example

client.delete_experiment_async(id="my_experiment")
async def delete_experiment_async(self, id: str) ‑> None

Delete an experiment (async)

Args

  • id: (str) Unique identifier of an experiment

Example

await client.delete_experiment_async(id="my_experiment")
def get_experiment(self, id: str) ‑> Experiment

Get an experiment.

Args

  • id: (str) Unique identifier of an experiment

Returns

Experiment

Example

experiment = client.get_experiment(id="my_experiment")
async def get_experiment_async(self, id: str) ‑> Experiment

Get an experiment (async).

Args

  • id: (str) Unique identifier of an experiment

Returns

Experiment

Example

experiment = await client.get_experiment_async(id="my_experiment")
def list_experiments(self) ‑> List[Experiment]

List experiments.

Returns

list(Experiment)

Example

experiments = client.list_experiments()
async def list_experiments_async(self) ‑> List[Experiment]

List experiments (async).

Returns

list(Experiment)

Example

experiments = await client.list_experiments_async()
def update_experiment(self, id: str, buckets: List[Dict[str, ]], bucketing_seed: str = '') ‑> Experiment

Update an experiment.

Args

  • id: (str) Unique identifier of an experiment
  • buckets: (list(dict)) Bucket list in an experiment
  • bucketing_seed: (optional) (str) Random seed for bucketing (default: "")

Returns

Experiment

Example

experiment = client.update_experiment(
    id="my_experiment", buckets=[{"name": "a", "ratio": 90}, {"name": "c", "ratio": 10}], bucketing_seed="200"
)
async def update_experiment_async(self, id: str, buckets: List[Dict[str, ]], bucketing_seed: str = '') ‑> Experiment

Update an experiment (async).

Args

  • id: (str) Unique identifier of an experiment
  • buckets: (list(dict)) Bucket list in an experiment
  • bucketing_seed: (optional) (str) Random seed for bucketing (default: "")

Returns

Experiment

Example

experiment = await client.update_experiment_async(
    id="my_experiment", buckets=[{"name": "a", "ratio": 90}, {"name": "c", "ratio": 10}], bucketing_seed="200"
)
def update_experiment_partial(self, id: str, buckets: List[Dict[str, ]] = None, bucketing_seed: str = None) ‑> Experiment

Partially update an experiment.

Args

  • id: (str) Unique identifier of an experiment
  • buckets: (optional) (list(dict)) Bucket list in an experiment
  • bucketing_seed: (optional) (str) Random seed for bucketing

Returns

Experiment

Example

experiment = client.update_experiment_partial(
    id="my_experiment", buckets=[{"name": "a", "ratio": 50}, {"name": "b", "ratio": 50}]
)
async def update_experiment_partial_async(self, id: str, buckets: List[Dict[str, ]] = None, bucketing_seed: str = None) ‑> Experiment

Partially update an experiment (async).

Args

  • id: (str) Unique identifier of an experiment
  • buckets: (optional) (list(dict)) Bucket list in an experiment
  • bucketing_seed: (optional) (str) Random seed for bucketing

Returns

Experiment

Example

experiment = await client.update_experiment_partial_async(
    id="my_experiment", buckets=[{"name": "a", "ratio": 50}, {"name": "b", "ratio": 50}]
)
class RecommendationClient (config: Config)

Client for BAP Recommendation API (/v1/recommend/).

Example

import bap

client = bap.client("recommend")

Ancestors

Methods

def recommend(self, id: str, channel_id: str, params: Dict[str, Any] = {}, headers: Dict[str, Any] = {}) ‑> Dict[str, Any]

Get recommended items for the target.

Args

  • id: (str) Unique identifier of a recommendation target
  • channel_id: (str) Unique identifier of a channel
  • params: (optional) (dict) Additional parameters for recommendation
  • headers: (optional) (dict) Headers being sent to API server

Returns

dict

Example

result = client.recommend(id="a", channel_id="my_channel", params={"first_param": 100, "second_param": "value"}
async def recommend_async(self, id: str, channel_id: str, params: Dict[str, Any] = {}, headers: Dict[str, Any] = {}) ‑> Dict[str, Any]

Get recommended items for the target (async).

Args

  • id: (str) Unique identifier of a recommendation target
  • channel_id: (str) Unique identifier of a channel
  • params: (optional) (dict) Additional parameters for recommendation
  • headers: (optional) (dict) Headers being sent to API server

Returns

dict

Example

result = await client.recommend_async(id="a", channel_id="my_channel", params={"first_param": 100, "second_param": "value"}