par Mark Tse, développeur de logiciels, Développement de plate-forme D2L
publié initialement le 24 mars 2018
Dans cet article, nous examinerons comment maintenir une base de données à jour avec les extraits quotidiens des ensembles de données Brightspace (BDS).
Introduction
L' publication de BDS permet aux administrateurs d’accéder à des données brutes qui peuvent être agrégées et filtrées selon leurs besoins. Ces ensembles de données sont actualisés quotidiennement, de sorte que les administrateurs doivent revenir quotidiennement à Brightspace pour obtenir les dernières données. Cet article vous aidera à commencer à automatiser ce processus, afin que vous puissiez passer plus de temps à travailler avec les données et à générer les rapports que vous souhaitez, et moins de temps à maintenir une base de données à jour.
Le code source de cet article est disponible sur GitHub.
Vue d’ensemble du flux de travail
Cet exemple de code utilise PostgreSQL et Python 3. Cependant, il est possible de faire ce flux de travail avec n’importe quelle base de données et langage de programmation.
Prérequis
- BDS activé sur Brightspace et programmé pour s’exécuter quotidiennement
- Enregistrement d’une demande OAuth 2.0 qui permet d’actualiser les jetons
- Un jeton d’actualisation initial
- Python 3.6, avec des bibliothèques dépendantes installées (
pip install -r requirements.txt) - Instance PostgreSQL (exemple testé par rapport à la version 9.6.2) et schéma configuré en exécutant les fichiers de schéma dans
schéma/tables - Un
config.json dans le même dossier que le script, avec les éléments suivants :
{ « bspace_url » : « https:/// »,
« client_id » : « »,
« client_secret » : « »,
« refresh_token » : « »,
« dbhost » : « »,
« dbname » : « »,
« dbuser » : « »,
« dbpassword » : « "}
Remarque : les autorisations de ce fichier de configuration doivent être définies de sorte qu’il ne soit lisible que par l’utilisateur exécutant le script, car il contient des informations sensibles.
Obtention d’un jeton d’accès OAuth 2.0
Tout d’abord, nous utilisons l’ID client et le secret de l’enregistrement de l’application OAuth 2.0 ainsi que le jeton d’actualisation que nous avons obtenu à l’étape précédente pour obtenir un jeton d’accès.
CONFIG_LOCATION = 'config.json'
AUTH_SERVICE = 'https://auth.brightspace.com/'
...
def get_config() :
avec open(CONFIG_LOCATION, 'r') comme f :
retour json.load(f)
def trade_in_refresh_token(config) :
# https://tools.ietf.org/html/rfc6749#section-6
réponse = requests.post(
'{}/core/connect/token'.format(AUTH_SERVICE),
# Content-Type 'application/x-www-form-urlencoded'
data={
« grant_type » : « refresh_token »,
'refresh_token' : config['refresh_token'],
'scope' : 'core :*:*'
},
auth=HTTPBasicAuth(config['client_id'], config['client_secret'])
)
si response.status_code != 200 :
logger.error('Code d’état : %s ; contenu : %s', response.status_code, response.text)
response.raise_for_status()
retour response.json()
def put_config(config) :
avec open(CONFIG_LOCATION, 'w') comme f :
json.dump(config, f, sort_keys=True)
config = get_config()
token_response = trade_in_refresh_token(config)
# Stockez le nouveau jeton d’actualisation pour obtenir un nouveau jeton d’accès next runconfig['refresh_token'] = token_response['refresh_token']put_config(config)
Les jetons d’actualisation sont à usage unique. Cependant, nous obtenons un nouveau jeton d’actualisation avec la réponse du jeton d’accès, nous stockons donc le nouveau jeton d’actualisation dans config.json pour la prochaine exécution.
Téléchargement des ensembles de données
Les ensembles de données qui nous intéressent sont définis en haut du script :
DATA_SET_METADATA = [
DataSetMetadata(
plugin='07a9e561-e22f-4e82-8dd6-7bfb14c91776',
table='org_units'
),
DataSetMetadata(
plugin='793668a8-2c58-4e5e-b263-412d28d5703f',
table='grade_objects'
),
DataSetMetadata(
plugin='1d6d722e-b572-456f-97c1-d526570daa6b',
table='utilisateurs'
),
DataSetMetadata(
plugin='9d8a96b4-8145-416d-bd18-11402bc58f8d',
table='grade_results'
)]
Pour chaque ensemble de données, nous appelons Brightspace pour obtenir l’ensemble de données zippé :
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
)
headers = {'Authorization' : 'Bearer {}'.format(token_response['access_token'])}
response = requests.get(endpoint, headers=headers)
Extraction du fichier CSV
Pour extraire l’ensemble de données, nous chargeons l’archive zip en mémoire. Les flux sont utilisés pour éviter de charger l’ensemble du CSV décompressé dans la mémoire à la fois.
Il ne devrait y avoir qu’un seul fichier dans l’archive.
avec io. BytesIO(response.content) comme response_stream :
avec zipfile. ZipFile(response_stream) en tant que zipped_data_set :
fichiers = zipped_data_set.namelist()
assert len(files) == 1
csv_name = fichiers[0]
avec zipped_data_set.open(csv_name) comme csv_data :
update_db(db_conn_params, tableau, csv_data)
Mise à jour de la base de données
Pour mettre à jour la base de données, nous chargeons d’abord l’ensemble des données dans une table temporaire :
cur.execute(
'''
CREATE TEMP TABLE tmp_{table} AS
SÉLECTIONNER *
FROM {table}
LIMITE 0 ;
'''
.format(table=table)
)
cur.copy_expert(
'''
COPY tmp_{table}
DE STDIN
AVEC (FORMAT CSV, EN-TÊTE) ;
'''
.format(table=table),
csv_data
)
Remarque : la syntaxe ici est pour PostgreSQL, et peut différer selon le moteur SQL que vous utilisez.
Nous effectuons un upsert (insérer ou mettre à jour) à la table principale en chargeant la requête à l’intérieur schéma/upserts et son exécution. Cette requête upsert lira toutes les données de l' TEMP et mettre à jour ou insérer dans le tableau principal en fonction de l' CLÉ PRIMAIRE.
upsert_query_file = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
« schéma »,
« upserts »,
'upsert_{table}.sql'.format(table=table)
)
avec open(upsert_query_file) comme upsert_query :
cur.execute(upsert_query.read())
Voici un exemple de requête upsert pour PostgreSQL :
INSÉRER DANS LES utilisateurs
SÉLECTIONNER
user_id,
user_name,
org_defined_id,
first_name,
middle_name,
last_name,
is_active,
Organisation,
internal_email,
external_email,
signup_date
DE tmp_users
SUR LE CONFLIT SUR LA CONTRAINTE users_pkey
DÉFINIR LA MISE À JOUR
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,
organisation = EXCLUDED.organization,
internal_email = EXCLUDED.internal_email,
external_email = EXCLUDED.external_email,
signup_date = EXCLUDED.signup_date
D’autres moteurs SQL peuvent différer légèrement en offrant FUSIONNER À la place de SUR LE CONFLIT.
Enfin, nous nettoyons en laissant tomber le TEMP table et validation de la transaction :
cur.execute('DROP TABLE tmp_{table}'.format(table=table))
...
conn.commit()
Travail planifié
Ce script est réexpédable et peut être déclenché quotidiennement. Parallèlement à la tâche planifiée BDS, une exécution planifiée de ce script maintiendra la base de données à jour avec les dernières données.
Conclusion
Nous avons montré comment mettre à jour par programmation une base de données avec un ensemble donné d’extraits BDS quotidiens. Ces étapes suppriment la nécessité de mettre à jour manuellement la base de données à mesure que les extraits sont mis à jour, ce qui vous donne plus de temps pour vous concentrer sur l’analyse des données.