bayesian_ab_testing.api

This module contains an example API ready to go.

get_arm

handles GET at /arm

Gets arm information, or multiple if id = -1

Parameters:

Name	Type	Description	Default
id	int	id of arm. Defaults to -1.	-1

Returns:

Name	Type	Description
dict	dict	arm info

Source code in bayesian_ab_testing\api\api.py

@app.get("/arm")
async def get_arm(id: int = -1) -> dict:
    """Gets arm information, or multiple if id = -1

    Args:
        id (int, optional): id of arm. Defaults to -1.

    Returns:
        dict: arm info
    """    
    with SqlHandler("DimArm") as dim_arm:
        if id < 0:
            res = dim_arm.from_sql_to_pandas().to_dict()

            return {k:{id:v[id]} if id < len(v) and id >= 0 else v for k, v in res.items()} if id is not None else res
        else:
            res = dim_arm.select_one(id)

    return res

add_arm

handles POST at /arm

Adds an arm

Parameters:

Name	Type	Description	Default
customer_id	int	identifier	required
type	str	type	required
reward	float	Amount of reward per trigger	required

Returns:

Name	Type	Description
dict	dict	Arm as a dict

Source code in bayesian_ab_testing\api\api.py

@app.post("/arm")
async def add_arm(customer_id: int, type: str, reward: float) -> dict:
    """Adds an arm

    Args:
        customer_id (int): identifier
        type (str): type
        reward (float): Amount of reward per trigger

    Returns:
        dict: Arm as a dict
    """    
    id = SqlHandler("DimArm").get_next_id()
    ThompsonArm(id, cnxn, type=type, customer_id=customer_id, reward=reward)

    return {"arm_id": id, "type": type, "reward": reward, "active": 1}

mod_arm

handles PUT at /arm/type

Modifies the type of the arm

Parameters:

Name	Type	Description	Default
id	int	arm identifier	required
type	Union[str, None]	type	required

Returns:

Name	Type	Description
str	str	"ok" if ok

Source code in bayesian_ab_testing\api\api.py

@app.put("/arm/type")
async def mod_arm(id: int, type: Union[str, None]) -> str:
    """Modifies the type of the arm

    Args:
        id (int): arm identifier
        type (Union[str, None]): type

    Returns:
        str: "ok" if ok
    """    
    try:
        ThompsonArm(id).change_type(type)
    except ValueError as e:
        return "ValueError: " + str(e)
    return "ok"

toggle_arm

handles PUT at /arm

Turns an arm on/off

Parameters:

Name	Type	Description	Default
id	int	arm id	required

Returns:

Name	Type	Description
str	str	ok or error message

Source code in bayesian_ab_testing\api\api.py

@app.put("/arm")
async def toggle_arm(id: int) -> str:
    """Turns an arm on/off

    Args:
        id (int): arm id

    Returns:
        str: ok or error message
    """

    try:
        ThompsonArm(id, cnxn).toggle_active()
    except ValueError as e:
        logging.error(e)
        return "ValueError: " + str(e)

    return "ok"

sample_arm

handles GET at /sample

Samples from all arms and choses the best one

Parameters:

Name	Type	Description	Default
customer_id	int	Customer id	required
information	Union[str, None]	Additional client data	required

Returns:

Name	Type	Description
dict	dict	a serve object as a dict

Source code in bayesian_ab_testing\api\api.py

@app.get("/sample")
async def sample_arm(customer_id: int, information: Union[str, None]) -> dict:
    """Samples from all arms and choses the best one

    Args:
        customer_id (int): Customer id
        information (Union[str, None]): Additional client data

    Returns:
        dict: a serve object as a dict
    """    
    algo = ThompsonAlgo(cnxn, customer_id)
    serve = algo.get_best_arm(information)

    return serve

log_result

handles PUT at /sample

Logs a trigger

Parameters:

Name	Type	Description	Default
customer_id	int	Customer identifier	required
arm_id	int	Arm identifier	required
serve_id	int	Serve identifier	required

Returns:

Name	Type	Description
str	str	"ok" of ok

Source code in bayesian_ab_testing\api\api.py

@app.put("/sample")
async def log_result(customer_id, arm_id, serve_id) -> str:
    """Logs a trigger

    Args:
        customer_id (int): Customer identifier
        arm_id (int): Arm identifier
        serve_id (int): Serve identifier

    Returns:
        str: "ok" of ok
    """    
    algo = ThompsonAlgo(cnxn, customer_id)
    [arm for arm in algo.arms if arm.id == int(arm_id)][0].log_trigger(serve_id)
    return "ok"

stats

handles GET at /stats

Summary statistics and state of an arm

Parameters:

Name	Type	Description	Default
arm_id	int	arm identifier	required

Returns:

Name	Type	Description
dict	dict	Dict with aggregated values and state

Source code in bayesian_ab_testing\api\api.py

@app.get("/stats")
async def stats(arm_id: int) -> dict:
    """Summary statistics and state of an arm

    Args:
        arm_id (int): arm identifier

    Returns:
        dict: Dict with aggregated values and state
    """    
    agr = SqlHandler("AggregateResult")
    res = agr.select_one(arm_id)
    return res

add_client

handles POST at /client

Adds a client to the db

Parameters:

Name	Type	Description	Default
name	str	Name	required
location	str	location	required
contact	str	contact info	required

Returns:

Name	Type	Description
dict	dict	customer as a dict

Source code in bayesian_ab_testing\api\api.py

@app.post("/client")
async def add_client(name:str, location: str, contact: str) -> dict:
    """Adds a client to the db

    Args:
        name (str): Name
        location (str): location
        contact (str): contact info

    Returns:
        dict: customer as a dict
    """    
    cust = SqlHandler("DimCustomer")
    id = cust.get_next_id()
    cust_dict = dict(customer_id=id, name=name, location=location, contact=contact)
    cust.insert_one(**cust_dict)

    return cust_dict