Projects Table

projects/projects.py¶

ProjectsDB ¶

ProjectsDB(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
`ProjectsDB`	An instance of this class

Source code in tm_admin/projects/projects.py

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

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

    Returns:
        (ProjectsDB): An instance of this class
    """
    self.pg = None
    self.profile = ProjectsTable()
    self.types = dir(tm_admin.types_tm)
    super().__init__('projects')

mergeAuxTables `async` ¶

mergeAuxTables(inuri, outuri)

Merge more tables from TM into the unified projects table.

Parameters:

Name	Type	Description	Default
`inuri`	`str`	The input database	required
`outuri`	`str`	The output database	required

Source code in tm_admin/projects/projects.py

async def mergeAuxTables(self,
                         inuri: str,
                         outuri: str,
                         ):
    """
    Merge more tables from TM into the unified projects table.

    Args:
        inuri (str): The input database
        outuri (str): The output database
    """
    await self.connect(outuri)

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

    await self.mergeAllowed(inpg)

    await self.mergeTeams(inpg)

    await self.mergeInfo(inpg)

    await self.mergeChat(inpg)

    await self.mergeInterests(inpg)

    await self.mergePriorities(inpg)

updateThread `async` ¶

updateThread(queries, db)

Thread to handle importing data

Parameters:

Name	Type	Description	Default
`queries`	`list`	The list of SQL queries to execute	required
`db`	`PostgresClient`	A database connection	required

Source code in tm_admin/projects/projects.py

async def updateThread(
    queries: list,
    db: PostgresClient,
):
    """Thread to handle importing data

    Args:
        queries (list): The list of SQL queries to execute
        db (PostgresClient): A database connection
    """
    pbar = tqdm.tqdm(queries)
    # for sql in queries:
    for sql in pbar:
        # print(sql)
        result = await db.execute(sql)

    return True

main `async` ¶

main()

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

Source code in tm_admin/projects/projects.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="Database URI")
    parser.add_argument("-o", "--outuri", default='localhost/tm_admin',
                            help="Database URI")
    parser.add_argument("-b", "--boundary", help="The project AOI")

    # 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,
    )

    proj = ProjectsDB(args.outuri)
    # user.resetSequence()
    #all = proj.getAll()

    timer = Timer(initial_text=f"Merging all other tables...",
                      text="Importing took {seconds:.0f}s",
                      logger=log.debug,
                    )
    timer.start()
    await proj.mergeAuxTables(args.inuri, args.outuri)
    timer.stop()

options: show_source: false heading_level: 3

projects/projects_class.py¶

options: show_source: false heading_level: 3

projects/api.py¶

ProjectsAPI ¶

ProjectsAPI()

Bases: PGSupport

Returns:

Type	Description
`ProjectsAPI`	An instance of this class

Source code in tm_admin/projects/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:
        (ProjectsAPI): An instance of this class
    """
    self.allowed_roles = [
        Teamrole.TEAM_MAPPER,
        Teamrole.TEAM_VALIDATOR,
        Teamrole.TEAM_MANAGER,
    ]
    from tm_admin.users.api import UsersAPI
    self.users = None
    self.tasks = None
    super().__init__("projects")

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/projects/api.py

async def initialize(self,
                     inuri: str,
                  ) -> None:
    """
    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("projects")
    self.users = tm_admin.users.api.UsersAPI()
    self.tasks = tm_admin.tasks.api.TasksAPI()

getByID `async` ¶

getByID(project_id)

Get all the information for a project using it's ID

Parameters:

Name	Type	Description	Default
`project_id`	`int`	The team to get the data for	required

Returns:

Type	Description
`dict`	the project information

Source code in tm_admin/projects/api.py

async def getByID(self,
                 project_id: int,
                ):
    """
    Get all the information for a project using it's ID

    Args:
        project_id (int): The team to get the data for

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

getTeamRole `async` ¶

getTeamRole(project_id, team_id)

Parameters:

Name	Type	Description	Default
`project_id`	`id`	The project ID	required
`team_id`	`id`	The team ID	required

Returns:

Type	Description
`Teamrole`	The role of this team in this project

Source code in tm_admin/projects/api.py

async def getTeamRole(self,
                    project_id: int,
                    team_id: int,
                    ):
    """
    Args:
        project_id (id): The project ID
        team_id (id): The team ID

    Returns:
        (Teamrole): The role of this team in this project
    """
    # log.warning(f"getProjectByTeam(): unimplemented!")
    # sql = f"SELECT FROM projects WHERE project_id={project_id}"
    # where = [{'teams': {"team_id": team_id, "project_id": project_id}}]
    #data = await self.getColumns(['id', 'teams'], where)
    # The role is in a list of dicts in a jsonb column.

    sql = f"SELECT jsonb_path_query(teams, '$.teams[*] ? (@.team_id[*] == {team_id})') AS results FROM projects WHERE id = {project_id};"
    results = await self.execute(sql)

    # There should only be one item in the results. Since it's a jsonb column
    # the data is returned as a string. In the string is an enum value, which
    # gets converted to the actual enum for Teamrole.
    if len(results) > 0:
        if results[0]['results'][0] == '{':
            tmp1 = eval(results[0]['results'])
            tmp2 = f"Teamrole.{tmp1['role']}"
            role = eval(tmp2)
            return role

    # we should never get here, but you never know...
        return None

getByName `async` ¶

getByName(name)

Get all the information for a project using the name

Parameters:

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

Returns:

Type	Description
`dict`	the project information

Source code in tm_admin/projects/api.py

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

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

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

changeStatus `async` ¶

changeStatus(project_id, status)

Manage the status of a task. This would be locking or unlocking, validation status, etc...

Parameters:

Name	Type	Description	Default
`project_id`	`int`	The project ID to change	required
`status`	`ProjectStatus`	The status to change to	required

Returns:

Type	Description
`bool`	Whether locking/unlocking the task was sucessful

Source code in tm_admin/projects/api.py

async def changeStatus(self,
                    project_id: int,
                    status: Projectstatus,
                    ):
    """
    Manage the status of a task. This would be locking or unlocking,
    validation status, etc...

    Args:
        project_id (int): The project ID to change
        status (ProjectStatus): The status to change to

    Returns:
        (bool): Whether locking/unlocking the task was sucessful
    """
    log.warning(f"changeStatus(): unimplemented!")

    return False

evaluateMappingPermissions `async` ¶

evaluateMappingPermissions(uid, pid, perm)

evaluate_mapping_permission()

Parameters:

Name	Type	Description	Default
`uid`	`int`	The user ID	required
`pid`	`int`	The project ID	required
`perm`	`Permissions`	The permission level to check against	required

Returns:

Type	Description
`bool`	If the user has the right permissions for this project

Source code in tm_admin/projects/api.py

async def evaluateMappingPermissions(self,
                               uid: int,
                               pid: int,
                               perm: Permissions,
                               ):
    """
    evaluate_mapping_permission()

    Args:
        uid (int): The user ID
        pid (int): The project ID
        perm (Permissions): The permission level to check against

    Returns:
        (bool): If the user has the right permissions for this project
    """
    teamperm = Mappingnotallowed('USER_NOT_TEAM_MEMBER')
    result = await project.getByWhere(f" ")

    # See if user is on a team with team permissions
    level = user.getColumn(uid, 'mapping_level')
    if level != Permissions() or Permissions():
        pass

    return False

permittedUser `async` ¶

permittedUser(project_id, user_id)

Is a user permitted to map on this project.

Parameters:

Name	Type	Description	Default
`user_id`	`int`	The user ID to lock	required
`project_id`	`int`	The project this task is in	required

Returns:

Type	Description
`Mappingnotallowed`	The results of checking permissions

Source code in tm_admin/projects/api.py

async def permittedUser(self,
                        project_id: int,
                        user_id: int,
                        ):
    """
    Is a user permitted to map on this project.

    Args:
        user_id (int): The user ID to lock
        project_id (int): The project this task is in

    Returns:
        (Mappingnotallowed): The results of checking permissions
    """

    # FIXME: is the user blocked ?

    # FIXME: see it the users is allowed to work on this project
    log.warning(f"permittedUser(): unimplemented!")
    result = await self.getColumns(["allowed_users"],
                                    {"project_id": project_id})

    # FIXME: Has user accepted license
    log.warning(f"permittedUser(): unimplemented!")
    result = await self.users.getColumns(["licenses"],
                                    {"user_id": user_id})
    if len(result) == 0:
        return Mappingnotallowed.USER_NOT_ACCEPTED_LICENSE

    # FIXME: Then check project status
    result = await self.getColumns(["status"],
                                    {"project_id": project_id})
    if len(result) == 0:
        return Mappingnotallowed.PROJECT_NOT_PUBLISHED

    # FIXME: Then see if task is already locked
    result = await self.users.getColumns(["task_status"],
                                    {"user_id": user_id})


    if user_id in result[0]:
        return Mappingnotallowed.USER_ALLOWED
    else:
        # FIXME: add errors
        pass

toggleFavorites `async` ¶

toggleFavorites(flag=None)

Add or remove this project favorites for a user.

Parameters:

Name	Type	Description	Default
`flag`	`bool`	The value to set to, otherwise defaults to flipping it.	`None`

Source code in tm_admin/projects/api.py

async def toggleFavorites(self,
                          flag: bool = None,
                          ):
    """
    Add or remove this project favorites for a user.

    Args:
        flag (bool): The value to set to, otherwise defaults to flipping it.
    """
    log.warning(f"toggleFavorites(): Unimplemented!")

    return False

toggleFeatures `async` ¶

toggleFeatures(flag=None)

Args:

Returns:

Source code in tm_admin/projects/api.py

async def toggleFeatures(self,
                          flag: bool = None,
                         ):
    """

    Args:


    Returns:

    """
    log.warning(f"toggleFeatures(): Unimplemented!")

    return False

unlockTasks `async` ¶

unlockTasks(project_id)

Args:

Returns:

Source code in tm_admin/projects/api.py

async def unlockTasks(self,
                          project_id: int,
                         ):
    """

    Args:


    Returns:

    """
    log.warning(f"toggleFeatures(): Unimplemented!")

    return False

getUserStats `async` ¶

getUserStats(user_id, project_id)

Args:

Returns:

Source code in tm_admin/projects/api.py

async def getUserStats(self,
                        user_id: int,
                        project_id: int,
                         ):
    """

    Args:


    Returns:

    """
    log.warning(f"getUserStats(): Unimplemented!")

    return False

getProjectStats `async` ¶

getProjectStats(project_id)

Args:

Returns:

Source code in tm_admin/projects/api.py

async def getProjectStats(self,
                         project_id: int,
                         ):
    """

    Args:


    Returns:

    """
    log.warning(f"getProjectStats(): Unimplemented!")

    return False

getAOI `async` ¶

getAOI(project_id)

Parameters:

Name	Type	Description	Default
`project_id`	`int`	The project to get the data for	required

Returns:

Source code in tm_admin/projects/api.py

async def getAOI(self,
                     project_id: int,
                     ):
    """

    Args:
        project_id (int): The project to get the data for

    Returns:

    """
    # log.warning(f"getAOI(): Unimplemented!")
    data = await self.getColumns(['geometry'],  {"id": project_id})

    # Convert the WKB to a Polygon.
    return wkb.loads(data[0]['geometry'])

getDailyContributions `async` ¶

getDailyContributions(project_id)

Parameters:

Name	Type	Description	Default
`project_id`	`int`	The project to get the data for	required

Returns:

Source code in tm_admin/projects/api.py

async def getDailyContributions(self,
                           project_id: int,
                            ):
    """

    Args:
        project_id (int): The project to get the data for


    Returns:

    """
    log.warning(f"getDailyContributions(): Unimplemented!")

    return False

getProjectSummary `async` ¶

getProjectSummary()

Args:

Returns:

Source code in tm_admin/projects/api.py

async def getProjectSummary(self):
    """

    Args:

    Returns:

    """
    log.warning(f"getProjectSummary(): Unimplemented!")

    return False

main `async` ¶

main()

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

Source code in tm_admin/projects/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

Projects Table

projects/projects.py¶

ProjectsDB ¶

mergeAuxTables async ¶

updateThread async ¶

main async ¶

projects/projects_class.py¶

projects/api.py¶

ProjectsAPI ¶

initialize async ¶

getByID async ¶

getTeamRole async ¶

getByName async ¶

changeStatus async ¶

evaluateMappingPermissions async ¶

permittedUser async ¶

toggleFavorites async ¶

toggleFeatures async ¶

unlockTasks async ¶

getUserStats async ¶

getProjectStats async ¶

getAOI async ¶

getDailyContributions async ¶

getProjectSummary async ¶

main async ¶

mergeAuxTables `async` ¶

updateThread `async` ¶

main `async` ¶

initialize `async` ¶

getByID `async` ¶

getTeamRole `async` ¶

getByName `async` ¶

changeStatus `async` ¶

evaluateMappingPermissions `async` ¶

permittedUser `async` ¶

toggleFavorites `async` ¶

toggleFeatures `async` ¶

unlockTasks `async` ¶

getUserStats `async` ¶

getProjectStats `async` ¶

getAOI `async` ¶

getDailyContributions `async` ¶

getProjectSummary `async` ¶

main `async` ¶