Skip to content

Messages Table

messages/messages.py

MessagesDB 

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

An instance of this class
Source code in tm_admin/messages/messages.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def __init__(self,
             dburi: str = "localhost/tm_admin",
            ):
    """
    A class to access the messages table.

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

    Returns:
        (MessagesDB): An instance of this class
    """
    self.pg = None
    self.profile = MessagesTable()
    self.types = dir(tm_admin.types_tm)
    super().__init__('messages')

getByFilter 

getByFilter(
    user_id,
    locale,
    page,
    page_size=10,
    sort_by=None,
    sort_direction=None,
    message_type=None,
    from_username=None,
    project=None,
    task_id=None,
    status=None,
)

Filter Args: msg (MessagesTable): The filter criteria

Returns:

Type Description
list

the messages that match the filter
Source code in tm_admin/messages/messages.py 
 81
 82
 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
def getByFilter(self,
                user_id: int,
                locale: str,
                page: int,
                page_size = 10,
                sort_by = None,
                sort_direction = None,
                message_type = None,
                from_username = None,
                project = None,
                task_id = None,
                status = None,
                ):
    """
    Filter
    Args:
        msg (MessagesTable): The filter criteria

    Returns:
        (list): the messages that match the filter
    """
    if sort_column is None:
        pass
    if project is not None:
        pass
    if task_id is not None:
        pass
    if status in ["read", "unread"]:
        pass
    if message_type:
        pass
    if from_username is not None:
        pass

main 

main()

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

Source code in tm_admin/messages/messages.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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")
    # 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,
    )

    message = MessagesDB(args.uri)

options: show_source: false heading_level: 3

messages/messages_class.py

options: show_source: false heading_level: 3

messages/api.py

MessagesAPI 

MessagesAPI()

Bases: PGSupport

Returns:

Type Description
MessagesAPI

An instance of this class
Source code in tm_admin/messages/api.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
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:
        (MessagesAPI): An instance of this class
    """
    # 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/messages/api.py 
75
76
77
78
79
80
81
82
83
84
85
86
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("messages")

getByID async 

getByID(message_id)

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

Parameters:

Name Type Description Default
message_id int

The message to get the data for

 required

Returns:

Type Description
dict

the message information
Source code in tm_admin/messages/api.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
async def getByID(self,
                 message_id: int,
                ):
    """
    Get all the information for an message using it's ID

    Args:
        message_id (int): The message to get the data for

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

getByName async 

getByName(name)

Get all the information for a message using the name

Parameters:

Name Type Description Default
name str

The message to get the data for

 required

Returns:

Type Description
dict

the message information
Source code in tm_admin/messages/api.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
async def getByName(self,
                    name: str,
                    ):
    """
    Get all the information for a message using the name

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

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

main async 

main()

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

Source code in tm_admin/messages/api.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
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