Campaigns Table

campaigns/campaigns.py¶

CampaignsDB ¶

CampaignsDB(dburi='localhost/tm_admin')

Bases: DBSupport

Parameters:

Name	Type	Description	Default
`dburi`	`str`	The URI string for the database connection	`'localhost/tm_admin'`

Returns:

Type	Description
`UsersDB`	An instance of this class

Source code in tm_admin/campaigns/campaigns.py

def __init__(self,
             dburi: str = "localhost/tm_admin",
            ):
    """
    A class to access the campaigns table.

    Args:
        dburi (str): The URI string for the database connection

    Returns:
        (UsersDB): An instance of this class
    """
    self.pg = None
    self.profile = UsersTable()
    self.types = dir(tm_admin.types_tm)
    super().__init__('campaigns')

mergeOrganizations `async` ¶

mergeOrganizations(inpg)

A method to merge the contents of the TM campaign_organizations into the campaigns table as an array.

Parameters:

Name	Type	Description	Default
`inpg`	`PostgresClient`	The input database	required

Source code in tm_admin/campaigns/campaigns.py

async def mergeOrganizations(self,
                    inpg: PostgresClient,
                    ):
    """
    A method to merge the contents of the TM campaign_organizations into
    the campaigns table as an array.

    Args:
        inpg (PostgresClient): The input database
    """
    # FIXME: this is a weird table, and only has 4 entries, none of which appear
    # to be in the other tables, so nothing updates.
    table = 'campaign_organisations'
    sql = f"SELECT * FROM {table} ORDER BY campaign_id"
    # print(sql)
    result = await inpg.execute(sql)

    data = dict()
    pbar = tqdm.tqdm(result)
    for record in result:
        sql = f" UPDATE campaigns SET organizations = organizations||{record['organisation_id']} WHERE id={record['campaign_id']};"
        # print(sql)
        await self.pg.execute(sql)

mergeProjects `async` ¶

mergeProjects(inpg)

A method to merge the contents of the TM campaign_projects into the campaigns table as an array.

Parameters:

Name	Type	Description	Default
`inpg`	`PostgresClient`	The input database	required

Source code in tm_admin/campaigns/campaigns.py

async def mergeProjects(self,
                    inpg: PostgresClient,
                    ):
    """
    A method to merge the contents of the TM campaign_projects into
    the campaigns table as an array.

    Args:
        inpg (PostgresClient): The input database
    """
    table = 'campaign_projects'
    sql = f"SELECT * FROM {table} ORDER BY campaign_id"
    # print(sql)
    result = await inpg.execute(sql)

    index = 0
    data = dict()
    pbar = tqdm.tqdm(result)

    for record in pbar:
        sql = f" UPDATE campaigns SET projects = projects||{record['project_id']} WHERE id={record['campaign_id']};"
        # print(sql)
        await self.pg.execute(sql)

main `async` ¶

main()

This main function lets this class be run standalone by a bash script.

Source code in tm_admin/campaigns/campaigns.py

async def main():
    """This main function lets this class be run standalone by a bash script."""
    parser = argparse.ArgumentParser()
    parser.add_argument("-v", "--verbose", nargs="?", const="0", help="verbose output")
    parser.add_argument("-i", "--inuri", default='localhost/tm4',
                            help="Input database URI")
    parser.add_argument("-o", "--outuri", default='localhost/tm_admin',
                            help="Output database URI")
    # parser.add_argument("-r", "--reset", help="Reset Sequences")
    args = parser.parse_args()

    # if len(argv) <= 1:
    #     parser.print_help()
    #     quit()

    # if verbose, dump to the terminal.
    log_level = os.getenv("LOG_LEVEL", default="INFO")
    if args.verbose is not None:
        log_level = logging.DEBUG

    logging.basicConfig(
        level=log_level,
        format=("%(asctime)s.%(msecs)03d [%(levelname)s] " "%(name)s | %(funcName)s:%(lineno)d | %(message)s"),
        datefmt="%y-%m-%d %H:%M:%S",
        stream=sys.stdout,
    )

    inpg = PostgresClient()
    await inpg.connect(args.inuri)

    camp = camp = CampaignsDB(args.inuri)
    await camp.connect(args.outuri)

    await camp.mergeProjects(inpg)
    await camp.mergeOrganizations(inpg)

options: show_source: false heading_level: 3

campaigns/campaigns_class.py¶

options: show_source: false heading_level: 3

campaigns/api.py¶

CampaignsAPI ¶

CampaignsAPI()

Bases: PGSupport

Returns:

Type	Description
`CampaignsAPI`	An instance of this class

Source code in tm_admin/campaigns/api.py

def __init__(self):
    """
    Create a class to handle the backend API calls, so the code can be shared
    between test cases and the actual code.

    Returns:
        (CampaignsAPI): An instance of this class
    """
    # self.allowed_roles = [
    #     Teamrole.TEAM_MAPPER,
    #     Teamrole.TEAM_VALIDATOR,
    #     Teamrole.TEAM_MANAGER,
    # ]
    # self.messagesdb = MessagesDB()
    # self.usersdb = UsersDB()
    # self.teamsdb = TeamsDB()
    super().__init__("campaigns")

initialize `async` ¶

initialize(inuri)

Connect to all tables for API endpoints that require accessing multiple tables.

Parameters:

Name	Type	Description	Default
`inuri`	`str`	The URI for the TM Admin output database	required

Source code in tm_admin/campaigns/api.py

async def initialize(self,
                  inuri: str,
                  ):
    """
    Connect to all tables for API endpoints that require
    accessing multiple tables.

    Args:
        inuri (str): The URI for the TM Admin output database
    """
    await self.connect(inuri)
    await self.getTypes("campaigns")

getByID `async` ¶

getByID(campaign_id)

Get all the information for an campaign using it's ID

Parameters:

Name	Type	Description	Default
`campaign_id`	`int`	The campaign to get the data for	required

Returns:

Type	Description
`dict`	the campaign information

Source code in tm_admin/campaigns/api.py

async def getByID(self,
                 campaign_id: int,
                ):
    """
    Get all the information for an campaign using it's ID

    Args:
        campaign_id (int): The campaign to get the data for

    Returns:
        (dict): the campaign information
    """
    # log.debug(f"--- getByID() ---")
    sql = f"SELECT * FROM campaigns WHERE id={campaign_id}"
    results = await self.execute(sql)
    return results

getByName `async` ¶

getByName(name)

Get all the information for a campaign using the name

Parameters:

Name	Type	Description	Default
`name`	`str`	The campaign to get the data for	required

Returns:

Type	Description
`dict`	the campaign information

Source code in tm_admin/campaigns/api.py

async def getByName(self,
                    name: str,
                    ):
    """
    Get all the information for a campaign using the name

    Args:
        name (str): The campaign to get the data for

    Returns:
        (dict): the campaign information
    """
    # log.debug(f"--- getByName() ---")
    sql = f"SELECT * FROM campaigns WHERE name='{name}'"
    results = await self.execute(sql)
    return results

create `async` ¶

create(campaign)

Create a campaign and add it to the database.

Parameters:

Name	Type	Description	Default
`campaign`	`CampaignsTable`	The team data	required

Returns:

Type	Description
`bool`	Whether the campaign got created

Source code in tm_admin/campaigns/api.py

async def create(self,
                 campaign: CampaignsTable,
                 ):
    """
    Create a campaign and add it to the database.

    Args:
        campaign (CampaignsTable): The team data

    Returns:
        (bool): Whether the campaign got created
    """
    # log.warning(f"create(): unimplemented!")
    result = await self.insertRecords([campaign])

    # The ID of the record that just got inserted is returned
    if result:
        return True

    return False

update `async` ¶

update(campaign)

Update a campaign that is already in the database.

Parameters:

Name	Type	Description	Default
`campaign`	`CampaignsTable`	The campaign data	required

Returns:

Type	Description
`bool`	Whether the campaign got updated

Source code in tm_admin/campaigns/api.py

async def update(self,
                 campaign: CampaignsTable,
                 ):
    """
    Update a campaign that is already in the database.

    Args:
        campaign (CampaignsTable): The campaign data

    Returns:
        (bool): Whether the campaign got updated
    """
    log.warning(f"update(): unimplemented!")

    return False

main `async` ¶

main()

This main function lets this class be run standalone by a bash script.

Source code in tm_admin/campaigns/api.py

async def main():
    """This main function lets this class be run standalone by a bash script."""
    parser = argparse.ArgumentParser()
    parser.add_argument("-v", "--verbose", nargs="?", const="0", help="verbose output")
    parser.add_argument("-u", "--uri", default='localhost/tm_admin', help="Database URI")

    args = parser.parse_args()

    # if verbose, dump to the terminal.
    log_level = os.getenv("LOG_LEVEL", default="INFO")
    if args.verbose is not None:
        log_level = logging.DEBUG

    logging.basicConfig(
        level=log_level,
        format=("%(asctime)s.%(msecs)03d [%(levelname)s] " "%(name)s | %(funcName)s:%(lineno)d | %(message)s"),
        datefmt="%y-%m-%d %H:%M:%S",
        stream=sys.stdout,
    )

options: show_source: false heading_level: 3

Last update: March 6, 2024

Campaigns Table

campaigns/campaigns.py¶

CampaignsDB ¶

mergeOrganizations async ¶

mergeProjects async ¶

main async ¶

campaigns/campaigns_class.py¶

campaigns/api.py¶

CampaignsAPI ¶

initialize async ¶

getByID async ¶

getByName async ¶

create async ¶

update async ¶

main async ¶

mergeOrganizations `async` ¶

mergeProjects `async` ¶

main `async` ¶

initialize `async` ¶

getByID `async` ¶

getByName `async` ¶

create `async` ¶

update `async` ¶

main `async` ¶