Database

Commonly used database utilities

Database utilities, for functions that you use all the time.

db.utils.get_team_by_name(dbf, gh_team_name)

Query team by github team name.

Can only return a single team. If there are no teams with that name, or there are multiple teams with that name, we raise an error.

Raises

LookupError if the calling user, user to add, or specified team cannot be found in the database

Raises

RuntimeError if more than one team has the specified team name

Return type

app.model.team.Team

Returns

Team if found

Parameters
db.utils.get_team_members(dbf, team)

Query users that are members of the given team.

Return type

typing.List[app.model.user.User]

Returns

Users that belong to the team

Parameters
db.utils.get_users_by_ghid(dbf, gh_ids)

Query users by github user id.

Return type

typing.List[app.model.user.User]

Returns

List of users if found

Parameters

Database Facade

class db.facade.DBFacade

A database facade that gives an overall API for any databases.

Currently, we plan on having DynamoDB, but other databases, such as MongoDB or Postgres are also being considered. Please use this class instead of db/dynamodb.py, because we might change the databases, but the facade would stay the same.

abstract bulk_retrieve(Model, ks)

Retrieve a list of models from the database.

Keys not found in the database will be skipped. Should be at least as fast as multiple calls to .retrieve.

Parameters
  • Model (typing.Type[~T]) – the actual class you want to retrieve

  • ks (typing.List[str]) – retrieve based on this key (or ID)

Return type

typing.List[~T]

Returns

a list of models Model

abstract delete(Model, k)

Remove an object from a table.

Parameters
  • Model (typing.Type[~T]) – table type to remove the object from

  • k (str) – ID or key of the object to remove (must be primary key)

abstract query(Model, params=[])

Query a table using a list of parameters.

Returns a list of Model that have all of the attributes specified in the parameters. Every item in parameters is a tuple, where the first element is the user attribute, and the second is the value.

Example:

ddb = DynamoDb(config)
users = ddb.query(User, [('platform', 'slack')])

If you try to query a table without any parameters, the function will return all objects of that table.:

teams = ddb.query(Team)

Attributes that are sets (e.g. team.member) would be treated differently. This function would check to see if the entry contains a certain element. You can specify multiple elements, but they must be in different parameters (one element per tuple).:

teams = ddb.query(Team, [('members', 'abc123'),
                         ('members', '231abc')])
Parameters
  • Model (typing.Type[~T]) – type of list elements you’d want

  • params (typing.List[typing.Tuple[str, str]]) – list of tuples to match

Return type

typing.List[~T]

Returns

a list of Model that fit the query parameters

abstract query_or(Model, params=[])

Query a table using a list of parameters.

Returns a list of Model that have one of the attributes specified in the parameters. Some might say that this is a union of the parameters. Every item in parameters is a tuple, where the first element is the user attribute, and the second is the value.

Example:

ddb = DynamoDb(config)
users = ddb.query_or(User, [('platform', 'slack')])

If you try to query a table without any parameters, the function will return all objects of that table.:

teams = ddb.query_or(Team)

Attributes that are sets (e.g. team.member) would be treated differently. This function would check to see if the entry contains a certain element. You can specify multiple elements, but they must be in different parameters (one element per tuple).:

teams = ddb.query_or(Team, [('members', 'abc123'),
                            ('members', '231abc')])

The above would get you the teams that contain either member abc123 or 231abc.

Parameters
  • Model (typing.Type[~T]) – type of list elements you’d want

  • params (typing.List[typing.Tuple[str, str]]) – list of tuples to match

Return type

typing.List[~T]

Returns

a list of Model that fit the query parameters

abstract retrieve(Model, k)

Retrieve a model from the database.

Parameters
  • Model (typing.Type[~T]) – the actual class you want to retrieve

  • k (str) – retrieve based on this key (or ID)

Raises

LookupError if key is not found

Return type

~T

Returns

a model Model if key is found

abstract store(obj)

Store object into the correct table.

Object can be of type app.model.User or app.model.Team.

Parameters

obj (~T) – Object to store in database

Return type

bool

Returns

True if object was stored, and false otherwise

DynamoDB

class db.dynamodb.DynamoDB(config)

Handles calls to database through API.

Please do not use this class, and instead use db.facade.DBFacade. This class only works on DynamoDB, and should not be used outside of the facade class.

Parameters

config (config.Config) –

class Const(config)

A bunch of static constants and functions.

Used to convert between Python objects and the DDB table names, object attributes and table column names.

Parameters

config (config.Config) –

__init__(config)

Initialize the constants.

Parameters

config (config.Config) –

get_key(table_name)

Get primary key of the table name.

Parameters
  • cls – the name of the table

  • table_name (str) –

Raises

TypeError if table does not exist

Return type

str

Returns

primary key of the table

get_set_attrs(table_name)

Get class attributes that are sets.

Parameters
  • cls – the table name

  • table_name (str) –

Raises

TypeError if table does not exist

Return type

typing.List[str]

Returns

set attributes

get_table_name(cls)

Convert class into corresponding table name.

Parameters

cls (typing.Type[~T]) – Either User or Team

Raises

TypeError if it is not either User or Team

Return type

str

Returns

table name string

__init__(config)

Initialize facade using DynamoDB settings.

To avoid local tests failure when the DynamoDb server is used, an environment variable config.aws_local is read.

if config.aws_local:
    # Connect to locally-run instance of DynamoDB
    pass
else:
    # Connect to remote instance of DynamoDB
    pass
Parameters

config (config.Config) – configuration used to initialize

bulk_retrieve(Model, ks)

Retrieve a list of models from the database.

Keys not found in the database will be skipped. Should be at least as fast as multiple calls to .retrieve.

Parameters
  • Model (typing.Type[~T]) – the actual class you want to retrieve

  • ks (typing.List[str]) – retrieve based on this key (or ID)

Return type

typing.List[~T]

Returns

a list of models Model

check_valid_table(table_name)

Check if table with table_name exists.

Parameters

table_name (str) – table identifier

Return type

bool

Returns

boolean value, true if table exists, false otherwise

delete(Model, k)

Remove an object from a table.

Parameters
  • Model (typing.Type[~T]) – table type to remove the object from

  • k (str) – ID or key of the object to remove (must be primary key)

query(Model, params=[])

Query a table using a list of parameters.

Returns a list of Model that have all of the attributes specified in the parameters. Every item in parameters is a tuple, where the first element is the user attribute, and the second is the value.

Example:

ddb = DynamoDb(config)
users = ddb.query(User, [('platform', 'slack')])

If you try to query a table without any parameters, the function will return all objects of that table.:

teams = ddb.query(Team)

Attributes that are sets (e.g. team.member) would be treated differently. This function would check to see if the entry contains a certain element. You can specify multiple elements, but they must be in different parameters (one element per tuple).:

teams = ddb.query(Team, [('members', 'abc123'),
                         ('members', '231abc')])
Parameters
  • Model (typing.Type[~T]) – type of list elements you’d want

  • params (typing.List[typing.Tuple[str, str]]) – list of tuples to match

Return type

typing.List[~T]

Returns

a list of Model that fit the query parameters

query_or(Model, params=[])

Query a table using a list of parameters.

Returns a list of Model that have one of the attributes specified in the parameters. Some might say that this is a union of the parameters. Every item in parameters is a tuple, where the first element is the user attribute, and the second is the value.

Example:

ddb = DynamoDb(config)
users = ddb.query_or(User, [('platform', 'slack')])

If you try to query a table without any parameters, the function will return all objects of that table.:

teams = ddb.query_or(Team)

Attributes that are sets (e.g. team.member) would be treated differently. This function would check to see if the entry contains a certain element. You can specify multiple elements, but they must be in different parameters (one element per tuple).:

teams = ddb.query_or(Team, [('members', 'abc123'),
                            ('members', '231abc')])

The above would get you the teams that contain either member abc123 or 231abc.

Parameters
  • Model (typing.Type[~T]) – type of list elements you’d want

  • params (typing.List[typing.Tuple[str, str]]) – list of tuples to match

Return type

typing.List[~T]

Returns

a list of Model that fit the query parameters

retrieve(Model, k)

Retrieve a model from the database.

Parameters
  • Model (typing.Type[~T]) – the actual class you want to retrieve

  • k (str) – retrieve based on this key (or ID)

Raises

LookupError if key is not found

Return type

~T

Returns

a model Model if key is found

store(obj)

Store object into the correct table.

Object can be of type app.model.User or app.model.Team.

Parameters

obj (~T) – Object to store in database

Return type

bool

Returns

True if object was stored, and false otherwise

MemoryDB

class tests.memorydb.MemoryDB(users=[], teams=[])

An in-memory database.

To be used only in testing. Do not attempt to use it in production. Used when a test requires a database, but when we aren’t specifically testing database functionalities.

Stored objects can be mutated by external references if you don’t drop the reference after storing.

Parameters
  • users (typing.List[app.model.user.User]) –

  • teams (typing.List[app.model.team.Team]) –

__init__(users=[], teams=[])

Initialize with lists of objects.

Parameters
  • users (typing.List[app.model.user.User]) – list of users to initialize the db

  • teams (typing.List[app.model.team.Team]) – list of teams to initialize the db

bulk_retrieve(Model, ks)

Retrieve a list of models from the database.

Keys not found in the database will be skipped. Should be at least as fast as multiple calls to .retrieve.

Parameters
  • Model (typing.Type[~T]) – the actual class you want to retrieve

  • ks (typing.List[str]) – retrieve based on this key (or ID)

Return type

typing.List[~T]

Returns

a list of models Model

delete(Model, k)

Remove an object from a table.

Parameters
  • Model (typing.Type[~T]) – table type to remove the object from

  • k (str) – ID or key of the object to remove (must be primary key)

query(Model, params=[])

Query a table using a list of parameters.

Returns a list of Model that have all of the attributes specified in the parameters. Every item in parameters is a tuple, where the first element is the user attribute, and the second is the value.

Example:

ddb = DynamoDb(config)
users = ddb.query(User, [('platform', 'slack')])

If you try to query a table without any parameters, the function will return all objects of that table.:

teams = ddb.query(Team)

Attributes that are sets (e.g. team.member) would be treated differently. This function would check to see if the entry contains a certain element. You can specify multiple elements, but they must be in different parameters (one element per tuple).:

teams = ddb.query(Team, [('members', 'abc123'),
                         ('members', '231abc')])
Parameters
  • Model (typing.Type[~T]) – type of list elements you’d want

  • params (typing.List[typing.Tuple[str, str]]) – list of tuples to match

Return type

typing.List[~T]

Returns

a list of Model that fit the query parameters

query_or(Model, params=[])

Query a table using a list of parameters.

Returns a list of Model that have one of the attributes specified in the parameters. Some might say that this is a union of the parameters. Every item in parameters is a tuple, where the first element is the user attribute, and the second is the value.

Example:

ddb = DynamoDb(config)
users = ddb.query_or(User, [('platform', 'slack')])

If you try to query a table without any parameters, the function will return all objects of that table.:

teams = ddb.query_or(Team)

Attributes that are sets (e.g. team.member) would be treated differently. This function would check to see if the entry contains a certain element. You can specify multiple elements, but they must be in different parameters (one element per tuple).:

teams = ddb.query_or(Team, [('members', 'abc123'),
                            ('members', '231abc')])

The above would get you the teams that contain either member abc123 or 231abc.

Parameters
  • Model (typing.Type[~T]) – type of list elements you’d want

  • params (typing.List[typing.Tuple[str, str]]) – list of tuples to match

Return type

typing.List[~T]

Returns

a list of Model that fit the query parameters

retrieve(Model, k)

Retrieve a model from the database.

Parameters
  • Model (typing.Type[~T]) – the actual class you want to retrieve

  • k (str) – retrieve based on this key (or ID)

Raises

LookupError if key is not found

Return type

~T

Returns

a model Model if key is found

store(obj)

Store object into the correct table.

Object can be of type app.model.User or app.model.Team.

Parameters

obj (~T) – Object to store in database

Return type

bool

Returns

True if object was stored, and false otherwise