por Mark Tse, desenvolvedor de software, desenvolvimento de plataforma D2L
publicado originalmente em 24 de março de 2018
Neste artigo, veremos como manter um banco de dados atualizado com as extrações diárias dos conjuntos de dados do Brightspace (BDS).
Introdução
O liberação do BDS permite que os administradores acessem dados brutos que podem ser agregados e filtrados de acordo com suas necessidades. Esses conjuntos de dados são atualizados diariamente, portanto, os administradores precisam retornar diariamente ao Brightspace para obter os dados mais recentes. Este artigo ajudará você a começar a automatizar esse processo, para que você possa gastar mais tempo trabalhando com os dados e gerando os relatórios desejados e menos tempo mantendo um banco de dados atualizado.
O código-fonte deste artigo está disponível em GitHub.
Visão geral do fluxo de trabalho
Este exemplo de código usa PostgreSQL e Python 3. No entanto, é possível fazer esse fluxo de trabalho com qualquer banco de dados e linguagem de programação.
Pré-requisitos
- BDS ativado no Brightspace e programado para ser executado diariamente
- Registro de um aplicativo OAuth 2.0 que permite tokens de atualização
- Ano token de atualização inicial
- Python 3.6, com bibliotecas dependentes instaladas (
pip instalar -r requirements.txt) - Instância do PostgreSQL (amostra testada em 9.6.2) e esquema configurado executando os arquivos de esquema em
esquema/tabelas - Um
config.json na mesma pasta do script, com o seguinte:
{ "bspace_url": "https:///",
"client_id": "",
"client_secret": "",
"refresh_token": "",
"dbhost": "",
"dbname": "",
"dbuser": "",
"dbpassword": ""}
Nota: as permissões deste arquivo de configuração devem ser definidas para que seja legível apenas pelo usuário que executa o script, pois contém informações confidenciais.
Obtendo um token de acesso OAuth 2.0
Primeiro, usamos a ID do cliente e o segredo do registro do aplicativo OAuth 2.0, bem como o token de atualização obtido na etapa anterior para obter um token de acesso.
CONFIG_LOCATION = 'config.json'
AUTH_SERVICE = 'https://auth.brightspace.com/'
...
def get_config():
com open(CONFIG_LOCATION, 'r') como f:
retornar json.load(f)
def trade_in_refresh_token(config):
# https://tools.ietf.org/html/rfc6749#section-6
resposta = requests.post(
'{}/core/connect/token'.format(AUTH_SERVICE),
# Tipo de conteúdo 'application/x-www-form-urlencoded'
dados={
«grant_type»: «refresh_token»,
'refresh_token': config['refresh_token'],
'escopo': 'núcleo:*:*'
},
auth=HTTPBasicAuth(config['client_id'], config['client_secret'])
)
se response.status_code != 200:
logger.error('Código de status: %s; conteúdo: %s', response.status_code, response.text)
response.raise_for_status()
retornar response.json()
def put_config(config):
com open(CONFIG_LOCATION, 'w') como f:
json.dump(config, f, sort_keys=True)
configuração = get_config()
token_response = trade_in_refresh_token(config)
# Armazene o novo token de atualização para obter um novo token de acesso next runconfig['refresh_token'] = token_response['refresh_token']put_config(config)
Os tokens de atualização são de uso único. No entanto, obtemos um novo token de atualização com a resposta do token de acesso, portanto, armazenamos o novo token de atualização em config.json para a próxima execução.
Fazendo download dos conjuntos de dados
Os conjuntos de dados que estamos interessados são definidos na parte superior do script:
DATA_SET_METADATA = [
DataSetMetadata(
plugin='07a9e561-e22f-4e82-8dd6-7bfb14c91776',
tabela='org_units'
),
DataSetMetadata(
plugin='793668a8-2c58-4e5e-b263-412d28d5703f',
tabela='grade_objects'
),
DataSetMetadata(
plugin='1d6d722e-b572-456f-97c1-d526570daa6b',
tabela='usuários'
),
DataSetMetadata(
plugin='9d8a96b4-8145-416d-bd18-11402bc58f8d',
tabela='grade_results'
)]
Para cada conjunto de dados, chamamos o Brightspace para obter o conjunto de dados compactado:
API_VERSION = '1,15'
...
endpoint = '{bspace_url}/d2l/api/lp/{lp_version}/dataExport/bds/download/{plugin_id}'.format(
bspace_url=config['bspace_url'],
lp_version = API_VERSION,
plugin_id=plugin
)
cabeçalhos = {'Autorização': 'Portador {}'.format(token_response['access_token'])}
resposta = requests.get(endpoint, cabeçalhos=cabeçalhos)
Extraindo o arquivo CSV
Para extrair o conjunto de dados, carregamos o arquivo zip na memória. Os fluxos são usados para evitar o carregamento de todo o CSV descompactado na memória de uma só vez.
Deve haver apenas um arquivo no arquivo.
com io. BytesIO(response.content) como response_stream:
com arquivo zip. ZipFile(response_stream) como zipped_data_set:
arquivos = zipped_data_set.namelist()
assert len(arquivos) == 1
csv_name = arquivos[0]
com zipped_data_set.open(csv_name) como csv_data:
update_db(db_conn_params, tabela, csv_data)
Atualizando o banco de dados
Para atualizar o banco de dados, primeiro carregamos todo o conjunto de dados em uma tabela temporária:
cur.execute(
'''
CRIAR TABELA TEMPORÁRIA tmp_{table} COMO
SELECIONAR*
DE {tabela}
LIMITE 0;
'''
.format(tabela=tabela)
)
cur.copy_expert(
'''
CÓPIA DE tmp_{tabela}
DE STDIN
COM (FORMATO CSV, CABEÇALHO);
'''
.format(tabela=tabela),
csv_data
)
Nota: a sintaxe aqui é para PostgreSQL e pode diferir dependendo de qual mecanismo SQL você está usando.
Realizamos um upsert (inserir ou atualizar) na tabela principal carregando a consulta dentro schema/upserts e executá-lo. Essa consulta upsert lerá todos os dados no diretório TEMP tabela e atualizar ou inserir na tabela principal com base no CHAVE PRIMÁRIA.
upsert_query_file = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
'esquema',
«upserts»,
'upsert_{tabela}.sql'.format(tabela=tabela)
)
com open(upsert_query_file) como upsert_query:
cur.execute(upsert_query.read())
Aqui está um exemplo de consulta upsert para PostgreSQL:
INSERIR NOS usuários
SELECIONAR
user_id,
user_name,
org_defined_id,
first_name,
middle_name,
last_name,
is_active,
Organização
internal_email,
external_email,
signup_date
DE tmp_users
SOBRE CONFLITO SOBRE RESTRIÇÃO users_pkey
CONJUNTO DE ATUALIZAÇÃO
user_name = EXCLUDED.user_name,
org_defined_id = EXCLUDED.org_defined_id,
first_name = EXCLUDED.first_name,
middle_name = EXCLUDED.middle_name,
last_name = EXCLUDED.last_name,
is_active = EXCLUDED.is_active,
organização = EXCLUDED.organization,
internal_email = EXCLUDED.internal_email,
external_email = EXCLUDED.external_email,
signup_date = EXCLUDED.signup_date
Outros mecanismos SQL podem diferir ligeiramente oferecendo FUNDIR Em vez de SOBRE CONFLITO.
Finalmente, limpamos descartando o TEMP tabela e confirmando a transação:
cur.execute('DROP TABLE tmp_{table}'.format(table=table))
...
conn.commit()
Trabalho agendado
Esse script pode ser executado novamente e pode ser disparado diariamente. Juntamente com a tarefa agendada do BDS, uma execução agendada desse script manterá o banco de dados atualizado com os dados mais recentes.
Conclusão
Mostramos como atualizar programaticamente um banco de dados com um determinado conjunto de extrações diárias do BDS. Essas etapas eliminam a necessidade de atualizar manualmente o banco de dados à medida que as extrações são atualizadas, fornecendo mais tempo para você se concentrar na análise dos dados.