Skip to content

Organizations Table

organizations.py

OrganizationsDB 

OrganizationsDB(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
OrganizationsDB

An instance of this class
Source code in tm_admin/organizations/organizations.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(self,
             dburi: str = "localhost/tm_admin",
            ):
    """
    A class to access the organizations table.

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

    Returns:
        (OrganizationsDB): An instance of this class
    """
    self.pg = None
    self.profile = OrganizationsTable()
    self.types = dir(tm_admin.types_tm)
    super().__init__('organizations')

mergeManagers async 

mergeManagers(inpg)

A method to merge the contents of the TM organisation_managers into the orgsanizations table as an array.

Parameters:

Name Type Description Default
inpg PostgresClient

The input database

 required
Source code in tm_admin/organizations/organizations.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
async def mergeManagers(self,
                    inpg: PostgresClient,
                    ):
    """
    A method to merge the contents of the TM organisation_managers into
    the orgsanizations 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 = 'organisation_managers'
    sql = f"SELECT * FROM {table} ORDER BY organisation_id"
    # print(sql)
    result = await inpg.execute(sql)

    data = dict()
    pbar = tqdm.tqdm(result)
    for record in result:
        sql = f" UPDATE organizations SET managers = managers||{record['user_id']} WHERE id={record['organisation_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/organizations/organizations.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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)

    org = OrganizationsDB()
    await org.connect(args.outuri)

    await org.mergeManagers(inpg)

options: show_source: false heading_level: 3

organizations_class.py

options: show_source: false heading_level: 3

organizations/api.py

OrganizationsAPI 

OrganizationsAPI()

Bases: PGSupport

Returns:

Type Description
OrganizationsAPI

An instance of this class
Source code in tm_admin/organizations/api.py 
53
54
55
56
57
58
59
60
61
62
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:
        (OrganizationsAPI): An instance of this class
    """
    self.orgdb = OrganizationsDB()
    super().__init__("organizations")

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/organizations/api.py 
64
65
66
67
68
69
70
71
72
73
74
75
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("organizations")

getByID async 

getByID(org_id)

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

Parameters:

Name Type Description Default
org_id int

The organization to get the data for

 required

Returns:

Type Description
dict

the organization information
Source code in tm_admin/organizations/api.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def getByID(self,
                 org_id: int,
                ):
    """
    Get all the information for an organization using it's ID

    Args:
        org_id (int): The organization to get the data for

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

getByName async 

getByName(name)

Get all the information for a organization using the name

Parameters:

Name Type Description Default
name str

The organization to get the data for

 required

Returns:

Type Description
dict

the organization information
Source code in tm_admin/organizations/api.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def getByName(self,
                    name: str,
                    ):
    """
    Get all the information for a organization using the name

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

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

getStats async 

getStats(org_id)

Args:

Returns:

Source code in tm_admin/organizations/api.py 
111
112
113
114
115
116
117
118
119
120
121
122
async def getStats(self,
                   org_id: int,
                   ):
    """

    Args:


    Returns:

    """
    log.warning(f"getStats(): unimplemented!")

validateName async 

validateName(name)

Args:

Returns:

Source code in tm_admin/organizations/api.py 
124
125
126
127
128
129
130
131
132
133
134
135
async def validateName(self,
                       name: str,
                       ):
    """

    Args:


    Returns:

    """
    log.warning(f"validateName(): unimplemented!")

validateUser async 

validateUser(name)

Args:

Returns:

Source code in tm_admin/organizations/api.py 
137
138
139
140
141
142
143
144
145
146
147
148
async def validateUser(self,
                       name: str,
                       ):
    """

    Args:


    Returns:

    """
    log.warning(f"validateName(): unimplemented!")

main async 

main()

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

Source code in tm_admin/organizations/api.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
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/testdata', 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