Users Table

users/users.py

This class is used to import the auxilary tables into the primary table

UsersDB

UsersDB(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/users/users.py

def __init__(self,
             dburi: str = "localhost/tm_admin",
            ):
    """
    A class to access the users 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__('users', dburi)
    super().__init__('users')

mergeInterests async

mergeInterests(inpg)

Merge the user_interests table from the Tasking Manager

Parameters:

Name	Type	Description	Default
inpg	PostgresClient	The input database	required

Source code in tm_admin/users/users.py

async def mergeInterests(self,
                         inpg: PostgresClient,
                         ):
    """
    Merge the user_interests table from the Tasking Manager

    Args:
        inpg (PostgresClient): The input database
    """
    table = 'user_interests'
    timer = Timer(initial_text="Merging user_interests table...",
                  text="merging liceneses table took {seconds:.0f}s",
                  logger=log.debug,
                )
    log.info(f"Merging interests table...")
    timer.start()
    sql = "SELECT * FROM user_interests ORDER BY user_id"
    result = await inpg.execute(sql)

    # FIXME: this SQL turns out to be painfully slow, and we can create the array
    # in python faster.
    # await self.copyTable(table, remote)
    # await self.renameTable(table)
    # sql = f"SELECT row_to_json({table}) as row FROM {table}"
    # Not all records in this table have data
    # sql = f"SELECT u.user_id,ARRAY(SELECT ARRAY(SELECT c.interest_id FROM {table} c WHERE c.user_id = u.user_id)) AS user_id FROM {table} u;"
    data = list()
    entry = dict()
    prev = None
    # Restructure the data into a list, so we can more easily chop the data
    # into multiple smaller pieces, one for each thread.
    for record in result:
        if prev == record['user_id']:
            entry[record['user_id']].append(record['interest_id'])
        else:
            if len(entry) != 0:
                data.append(entry)
            prev = record['user_id']
            entry = {record['user_id']: [record['interest_id']]}
    timer.stop()

    # await remote.pg.close()
    #pg = PostgresClient()
    entries = len(data)
    chunk = round(entries / cores)

    start = 0
    async with asyncio.TaskGroup() as tg:
        for block in range(0, entries, chunk):
            # for index in range(0, cores):
            outpg = PostgresClient()
            await outpg.connect('localhost/tm_admin')
            log.debug(f"Dispatching thread {block}:{block + chunk}")
            # await interestsThread(data, outpg)
            task = tg.create_task(interestsThread(data[block:block + chunk], outpg))

    return True

mergeLicenses async

mergeLicenses(inpg)

Merge data from the TM user_licenses table into TM Admin. The fastest way to do a bulk update of a table is by copying the remote database table into the local database, and then merging into a new temporary table and then renaming it.

Parameters:

Name	Type	Description	Default
inpg	PostgresClient	The input database	required

Source code in tm_admin/users/users.py

async def mergeLicenses(self,
                         inpg: PostgresClient,
                         ):
    """
    Merge data from the TM user_licenses table into TM Admin. The
    fastest way to do a bulk update of a table is by copying the
    remote database table into the local database, and then merging
    into a new temporary table and then renaming it.

    Args:
        inpg (PostgresClient): The input database
    """
    table = 'user_licenses'
    # log.info(f"Merging licenses table...")
    timer = Timer(initial_text="Merging user_licenses table...",
                  text="merging user_liceneses table took {seconds:.0f}s",
                  logger=log.debug,
                )
    timer.start()
    sql = "SELECT * FROM user_licenses ORDER BY user"
    data = await inpg.execute(sql)

    entries = len(data)
    chunk = round(entries / cores)

    start = 0
    async with asyncio.TaskGroup() as tg:
        for block in range(0, entries, chunk):
            # for index in range(0, cores):
            outpg = PostgresClient()
            await outpg.connect('localhost/tm_admin')
            log.debug(f"Dispatching thread {block}:{block + chunk}")
            # await licensesThread(data, outpg)
            task = tg.create_task(licensesThread(data[block:block + chunk], outpg))

    # self.columns = await inpg.getColumns(table)
    # print(f"COLUMNS: {self.columns}")

    # await self.copyTable(table, self.pg)
    # # log.warning(f"Merging tables can take considerable time...")

    # # cleanup old temporary tables
    # drop = ["DROP TABLE IF EXISTS users_bak",
    #         "DROP TABLE IF EXISTS foo"]
    # # result = await pg.pg.executemany(drop)
    # sql = f"DROP TABLE IF EXISTS users_bak"
    # result = await self.pg.execute(sql)
    # sql = f"DROP TABLE IF EXISTS user_licenses"
    # result = await self.pg.execute(sql)
    # sql = f"DROP TABLE IF EXISTS new_users"
    # result = await self.pg.execute(sql)

    # # We need to use DBLINK
    # sql = f"CREATE EXTENSION IF NOT EXISTS dblink;"
    # data = await self.pg.execute(sql)

    # dbuser = self.pg.dburi["dbuser"]
    # dbpass = self.pg.dburi["dbpass"]
    # sql = f"CREATE SERVER IF NOT EXISTS pg_rep_db FOREIGN DATA WRAPPER dblink_fdw  OPTIONS (dbname 'tm4');"
    # data = await self.pg.execute(sql)

    # sql = f"CREATE USER MAPPING IF NOT EXISTS FOR {dbuser} SERVER pg_rep_db OPTIONS ( user '{dbuser}', password '{dbpass}');"
    # result = await self.pg.execute(sql)

    # # Copy table from remote database so JOIN is faster when it's in the
    # # same database
    # log.warning(f"Copying a remote table is slow, but faster than remote access......")
    # # sql = f"SELECT * INTO user_licenses FROM dblink('pg_rep_db','SELECT * FROM user_licenses') AS user_licenses(user_id bigint, license int)"
    # # print(pg.dburi)
    # # print(sql)
    # # result = await pg.execute(sql)

    # # JOINing into a new table is much faster than doing an UPDATE
    # sql = f"SELECT users.*,ARRAY[user_licenses.license] INTO new_users FROM users JOIN user_licenses ON users.id=user_licenses.user_id"
    # result = await self.pg.execute(sql)

    # # self.renameTable(table, pg)
    # # sql = f"ALTER TABLE users RENAME TO users_bak;"
    # # result = await pg.execute(sql)
    # # sql = f"ALTER TABLE new_users RENAME TO users;"
    # # result = await pg.execute(sql)
    # # sql = f"DROP TABLE IF EXISTS user_licenses"
    # # result = await pg.execute(sql)
    timer.stop()

mergeAuxTables async

mergeAuxTables(inuri, outuri)

Merge more tables from TM into the unified users table.

Parameters:

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

Source code in tm_admin/users/users.py

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

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

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

    await self.mergeFavorites(inpg)

    await self.mergeInterests(inpg)

    result = await self.mergeLicenses(inpg)

interestsThread async

interestsThread(interests, db)

Thread to handle importing

Parameters:

Name	Type	Description	Default
interests	list	The list of records to import	required
db	PostgresClient	A database connection	required

Source code in tm_admin/users/users.py

async def interestsThread(
    interests: list,
    db: PostgresClient,
):
    """Thread to handle importing

    Args:
        interests (list): The list of records to import
        db (PostgresClient): A database connection
    """
    pbar = tqdm.tqdm(interests)
    for record in pbar:
        for id, array in record.items():
            sql = f" UPDATE users SET interests = interests||ARRAY{array} WHERE id={id}"
            # print(sql)
            result = await db.execute(sql)

    return True

favoritesThread async

favoritesThread(favorites, db)

Thread to handle importing favorites

Parameters:

Name	Type	Description	Default
favorites	list	The list of records to import	required
db	PostgresClient	A database connection	required

Source code in tm_admin/users/users.py

async def favoritesThread(
    favorites: list,
    db: PostgresClient,
):
    """Thread to handle importing favorites

    Args:
        favorites (list): The list of records to import
        db (PostgresClient): A database connection
    """
    data = dict()
    pbar = tqdm(favorites)
    for record in pbar:
        for id, array in record.items():
            sql = f" UPDATE users SET favorite_projects = ARRAY{projects} WHERE id={uid}"
            # print(sql)
            result = await db.execute(f"{sql};")

    return True

main async

main()

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

Source code in tm_admin/users/users.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,
    )

    user = UsersDB()
    # user.connect(args.uri)
    await user.mergeAuxTables(args.inuri, args.outuri)

options: show_source: false heading_level: 3

users/users_class.py

options: show_source: false heading_level: 3

users/api.py

UsersAPI

UsersAPI()

Bases: PGSupport

Source code in tm_admin/users/api.py

def __init__(self):
    self.projects = None
    # self.projects = projects.api.ProjectsAPI()
    self.cursor = None
    super().__init__("users")

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/users/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("users")

    self.tasks = tm_admin.tasks.api.TasksAPI()
    await self.tasks.initialize(inuri)

    self.projects = tm_admin.projects.api.ProjectsAPI()
    await self.projects.initialize(inuri)

getByID async

getByID(user_id)

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

Parameters:

Name	Type	Description	Default
user_id	int	The user to get the data for	required

Returns:

Type	Description
dict	the user information

Source code in tm_admin/users/api.py

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

    Args:
        user_id (int): The user to get the data for

    Returns:
        (dict): the user information
    """
    # log.debug(f"--- getByID() ---")
    result = await self.getColumns(['*'], {"id": user_id})
    return result

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/users/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 users WHERE name='{name}'"
    results = await self.execute(sql)
    return results

getRole async

getRole(user_id)

Get the role for a user.

Parameters:

Name	Type	Description	Default
user_id	int	The user's ID	required

Returns:

Type	Description
Userrole	The user's role

Source code in tm_admin/users/api.py

async def getRole(self,
              user_id: int,
              ):
    """
    Get the role for a user.

    Args:
        user_id (int): The user's ID

    Returns:
        (Userrole): The user's role
    """
    result = await self.getColumns(['role'], {"id": user_id})

    return eval(f"Userrole.{result[0]['role']}")

getBlocked async

getBlocked(user_id)

Get the role for a user.

Parameters:

Name	Type	Description	Default
user_id	int	The user's ID	required

Returns:

Type	Description
bool	If the user is blocked or not

Source code in tm_admin/users/api.py

async def getBlocked(self,
              user_id: int,
              ):
    """
    Get the role for a user.

    Args:
        user_id (int): The user's ID

    Returns:
        (bool): If the user is blocked or not
    """
    result = await self.getColumns(['role'], {"id": user_id})

    role = eval(f"Userrole.{result[0]['role']}")
    if role == Userrole.USER_READ_ONLY:
        return True
    else:
        return False

updateMappingLevel async

updateMappingLevel(id, level)

Update the mapping level for a user.

Parameters:

Name	Type	Description	Default
id	int	The users ID.	required
level	Mappinglevel	The new level.	required

Source code in tm_admin/users/api.py

async def updateMappingLevel(self,
               id: int,
               level: Mappinglevel,
               ):
    """
    Update the mapping level for a user.

    Args:
        id (int): The users ID.
        level (Mappinglevel): The new level.
    """
    mlevel = Mappinglevel(level)
    result = await self.updateColumn(id, {'mapping_level': mlevel.name})
    return result

updateExpert async

updateExpert(id, mode)

Toggle the expert mode for a user.

Parameters:

Name	Type	Description	Default
id	int	The users ID.	required
mode	bool	The new mode..	required

Source code in tm_admin/users/api.py

async def updateExpert(self,
               id: int,
               mode: bool,
               ):
    """
    Toggle the expert mode for a user.

    Args:
        id (int): The users ID.
        mode (bool): The new mode..
    """
    result = await self.updateColumn(id, {'expert_mode': mode})
    return result

getRegistered async

getRegistered(start, end)

Get all users registered in this timeframe.

Parameters:

Name	Type	Description	Default
start	datetime	The starting timestamp	required
end	datetime	The starting timestamp	required

Returns:

Type	Description
list	The users registered in this timeframe.

Source code in tm_admin/users/api.py

async def getRegistered(self,
                  start: datetime,
                  end: datetime,
                  ):
    """
    Get all users registered in this timeframe.

    Args:
        start (datetime): The starting timestamp
        end (datetime): The starting timestamp

    Returns:
        (list): The users registered in this timeframe.
    """

    where = f" date_registered > '{start}' and date_registered < '{end}'"

getFavoriteProjects async

getFavoriteProjects(user_id)

Get the data for a users favorite projects.

Parameters:

Name	Type	Description	Default
user_id	int	The user to get the favorites for	required

Returns:

Type	Description
list	A list of the projects data

Source code in tm_admin/users/api.py

async def getFavoriteProjects(self,
                              user_id: int,
                              ):
    """
    Get the data for a users favorite projects.

    Args:
        user_id (int): The user to get the favorites for

    Returns:
        (list): A list of the projects data
    """
    result = await self.getColumns({"favorite_projects"}, {"id": user_id})
    data = list()
    for project_id in result[0]['favorite_projects']:
        data.append(await self.projects.getByID(project_id))

    return data

getPagedUsers async

getPagedUsers(
    paged,
    count,
    username=None,
    user_id=None,
    role=None,
    level=None,
)

Get paged list of all usernames using either the user ID or a partial username.

Parameters:

Name	Type	Description	Default
paged	bool	Whether to page the results	required
count	int	The number of entries in the page	required
username	str	The partial user name	None
user_id	int	The user ID	None
role	Userrole	The user's role	None
level	Mappinglevel	The users mapping level	None

Returns:

Type	Description
list	The users matching the query

Source code in tm_admin/users/api.py

async def getPagedUsers(self,
                        paged: bool,
                        count: int,
                        username: str = None,
                        user_id: int = None,
                        role: Userrole = None,
                        level: Mappinglevel = None,
                        ):
    """
    Get paged list of all usernames using either the
    user ID or a partial username.

    Args:
        paged (bool): Whether to page the results
        count (int): The number of entries in the page
        username (str): The partial user name
        user_id (int): The user ID
        role (Userrole): The user's role
        level (Mappinglevel): The users mapping level

    Returns:
         (list): The users matching the query
    """

getFilterUsers async

getFilterUsers(
    username, page=10, project_id=None, close=False
)

" Get paged lists of users matching OpenStreetMap using either the user ID or a partial username. The cursor gets closed when the class destructs, or by specifying the close parameter.

Parameters:

Name	Type	Description	Default
username	str	The partial user name	required
page	int	How many records in each page	10
project_id	int	Optional project ID	None
close	bool	Whether to close the cursor to restart from the beginning.	False

Returns:

Type	Description
list	The users matching the query

Source code in tm_admin/users/api.py

async def getFilterUsers(self,
                         username: str,
                         page: int = 10,
                         project_id: int = None,
                         close: bool = False,
                         ):
    """"
    Get paged lists of users matching OpenStreetMap
    using either the user ID or a partial username.
    The cursor gets closed when the class destructs,
    or by specifying the close parameter.

    Args:
        username (str): The partial user name
        page (int): How many records in each page
        project_id (int): Optional project ID
        close (bool): Whether to close the cursor to restart
                      from the beginning.

    Returns:
        (list): The users matching the query
    """
    try:
        result = await self.execute(f"SELECT FROM pg_cursor WHERE name = 'user_cursor'")
        print(result)
    except:
        pass
    if project_id:
        self.cursor = f"DECLARE user_cursor CURSOR WITH HOLD FOR SELECT username AS username FROM users WHERE username ILIKE '%{username}%' AND  {project_id} = ANY (projects_mapped) ORDER BY username DESC NULLS LAST, username"
    else:
        self.cursor = f"DECLARE user_cursor CURSOR WITH HOLD FOR SELECT username AS users_username FROM users WHERE username ILIKE '%{username}%' ORDER BY username DESC NULLS LAST, username"
    print(self.cursor)

    await self.execute(self.cursor)
    sql = f"FETCH FORWARD {page} FROM user_cursor"
    return await self.execute(sql)

getLockedTasks async

getLockedTasks(username=None, user_id=None)

Gets any locked tasks on the project for the logged in user using either the user ID or a partial username.

Parameters:

Name	Type	Description	Default
username	str	The partial user name	None
user_id	int	The ID of the logged in user	None

Returns:

Type	Description
list	The task IDs this user has locked

Source code in tm_admin/users/api.py

async def getLockedTasks(self,
                         username: str = None,
                         user_id: int = None,
                         ):
    """
    Gets any locked tasks on the project for the logged
    in user using either the user ID or a partial username.

    Args:
        username (str): The partial user name
        user_id (int): The ID of the logged in user

    Returns:
        (list): The task IDs this user has locked
    """
    result = await self.getByID(user_id)

getStats async

getStats(username=None, user_id=None)

Get detailed statistics about a user using either the user ID or a partial username.

Parameters:

Name	Type	Description	Default
username	str	The partial user name	None
user_id	int	The ID of the logged in user	None

Returns:

Type	Description
UserStats	The statistics for this user

Source code in tm_admin/users/api.py

async def getStats(self,
                   username: str = None,
                   user_id: int = None,
                   ):
    """
    Get detailed statistics about a user using either
    the user ID or a partial username.

    Args:
        username (str): The partial user name
        user_id (int): The ID of the logged in user

    Returns:
        (UserStats): The statistics for this user
    """

getInterestsStats async

getInterestsStats(username=None, user_id=None)

Get rate of contributions from a user given their interests using either the user ID or a partial username.

Parameters:

Name	Type	Description	Default
username	str	The partial user name	None
user_id	int	The ID of the logged in user	None

Returns:

Type	Description
int	The rate of contributions

Source code in tm_admin/users/api.py

async def getInterestsStats(self,
                            username: str = None,
                            user_id: int = None,
                            ):
    """
    Get rate of contributions from a user given their
    interests using either the user ID or a partial username.

    Args:
        username (str): The partial user name
        user_id (int): The ID of the logged in user

    Returns:
        (int): The rate of contributions
    """

getAllStats async

getAllStats(start, end)

Get rate of contributions from a user given their interests using either the user ID or a partial username.

Parameters:

Name	Type	Description	Default
start	datetime	The starting timestamp	required
end	datetime	The ending timestamp	required

Returns:

Type	Description
list	A list of the stats for the users in this time frame

Source code in tm_admin/users/api.py

async def getAllStats(self,
                      start: datetime,
                      end: datetime,
                      ):
    """
    Get rate of contributions from a user given their
    interests using either the user ID or a partial username.

    Args:
        start (datetime): The starting timestamp
        end (datetime): The ending timestamp

    Returns:
        (list): A list of the stats for the users in this
                time frame
    """

getInteracts async

getInteracts(
    tstatus,
    pstatus,
    project_id,
    start,
    end,
    sort,
    page=10,
    username=None,
    user_id=None,
)

Get a list of tasks a user has interacted with using either the user ID or a partial username.

Sort criteria is action_date or project_id.

Parameters:

Name	Type	Description	Default
username	str	The partial user name	None
user_id	int	The ID of the logged in user	None
tstatus	Taskstatus	The status of the task	required
pstatus	Projectstatus	The status of the project	required
project_id	int	The project ID	required
start	datetime	The starting timestamp	required
end	datetime	The ending timestamp	required
sort	str	The column to use for sorting the results	required
page	int	How many records in each page	10

Returns:

Type	Description
list	FIXME! a list of task data ?

Source code in tm_admin/users/api.py

async def getInteracts(self,
                   tstatus: Taskstatus,
                   pstatus: Projectstatus,
                   project_id: int,
                   start: datetime,
                   end: datetime,
                   sort: str,
                   page: int = 10,
                   username: str = None,
                   user_id: int = None,
                   ):
    """
    Get a list of tasks a user has interacted with
    using either the user ID or a partial username.

    Sort criteria is action_date or project_id.

    Args:
        username (str): The partial user name
        user_id (int): The ID of the logged in user
        tstatus (Taskstatus): The status of the task
        pstatus (Projectstatus): The status of the project
        project_id (int): The project ID
        start (datetime): The starting timestamp
        end (datetime): The ending timestamp
        sort (str): The column to use for sorting the results
        page (int): How many records in each page

    Returns:
        (list): FIXME! a list of task data ?
    """

main async

main()

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

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