Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def connect_keen(self, settings):
if 'keen' in settings:
self.keen_client = KeenClient(**settings['keen'])
else:
self.keen_client = KeenClient()
def get_keen_activity():
client = KeenClient(
project_id=settings.KEEN['public']['project_id'],
read_key=settings.KEEN['public']['read_key'],
)
node_pageviews = client.count(
event_collection='pageviews',
timeframe='this_7_days',
group_by='node.id',
filters=[
{
'property_name': 'node.id',
'operator': 'exists',
'property_value': True
}
]
)
es_client = None
if dry_run:
print("Doing dry-run upload to Elastic search. Pass --for-reals to upload to Keen")
es_client = Elasticsearch()
try:
es_client.indices.delete(script_settings.ES_INDEX)
except Exception as exc:
print(exc)
pass
else:
keen_clients = {
'public': KeenClient(
project_id=settings.KEEN['public']['project_id'],
write_key=settings.KEEN['public']['write_key'],
),
'private': KeenClient(
project_id=settings.KEEN['private']['project_id'],
write_key=settings.KEEN['private']['write_key'],
)
}
tally = {}
seen = {}
try:
with open(utils.get_dir_for('load') + '/resume.log', 'r') as resume_log:
for seen_file in resume_log.readlines():
seen[seen_file.strip('\n')] = 1
except:
pass
batch_count = utils.get_batch_count() if batch_count is None else batch_count
def connect_keen(self, settings):
if 'keen' in settings:
self.keen_client = KeenClient(**settings['keen'])
else:
self.keen_client = KeenClient()