Created
August 19, 2019 07:51
-
-
Save brydavis/89e8c5d0a55730a014f9554147f1cc3c to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import boto3 | |
from boto3.dynamodb.conditions import Key, Attr | |
import csv | |
import json | |
# Get the service resource. | |
session = boto3.Session(profile_name="default") | |
dynamodb = session.resource('dynamodb', region_name='us-west-2') | |
def strip_non_ascii(string): | |
''' Returns the string without non ASCII characters''' | |
stripped = (c for c in string if 0 < ord(c) < 127) | |
return ''.join(stripped) | |
# Create the DynamoDB table. | |
def dynamo_create_table(table_name, key_schema, attribute_definitions): | |
try: | |
table = dynamodb.create_table( | |
TableName=table_name, | |
KeySchema=key_schema, | |
AttributeDefinitions=attribute_definitions, | |
ProvisionedThroughput={ | |
'ReadCapacityUnits': 5, | |
'WriteCapacityUnits': 5 | |
} | |
) | |
# Wait until the table exists. | |
table.meta.client.get_waiter('table_exists').wait(TableName=table_name) | |
print("table created") | |
return True | |
except Exception as e: | |
print(e) | |
return False | |
def dynamo_insert_one(table_name, item): | |
table = dynamodb.Table(table_name) | |
try: | |
table.put_item( | |
Item=item | |
) | |
return True | |
except Exception as e: | |
return False | |
def import_data(data_dir, *files): | |
for filepath in files: | |
collection_name = filepath.split(".")[0] | |
print("opening", "/".join([data_dir, filepath])) | |
with open("/".join([data_dir, filepath])) as file: | |
reader = csv.reader(file, delimiter=",") | |
header = False | |
for row in reader: | |
if not header: | |
header = [h.strip("\ufeff").strip("").strip() for h in row] | |
dynamo_create_table( | |
collection_name, | |
[ | |
{ | |
'AttributeName': header[0], | |
'KeyType': 'HASH' | |
} | |
], | |
[ | |
{ | |
'AttributeName': header[0], | |
'AttributeType': 'S' | |
}, | |
], | |
) | |
else: | |
data = {header[i]:v for i,v in enumerate(row)} | |
print(data) | |
try: | |
dynamo_insert_one(collection_name, data) | |
except Exception as e: | |
print(e) | |
print(data) | |
"""pip install Flask""" | |
from flask import Flask, request, jsonify | |
app = Flask(__name__) | |
@app.route('/') | |
def index(): | |
with open("src/index.html") as file: | |
return file.read() | |
@app.route('/all') | |
def all_data (): | |
return json.dumps({ | |
"product": dynamodb.Table("demo_products").scan()["Items"], | |
}) | |
@app.route('/select', methods=['POST']) | |
def select_data(): | |
if request.method == 'POST': #this block is only entered when the form is submitted | |
product_id = request.form["product_id"] | |
rental = dynamodb.Table("demo_rental") | |
response = rental.scan( | |
FilterExpression=Key('product_id').eq(product_id) | |
) | |
return json.dumps(response) | |
else: | |
return {} | |
if __name__ == "__main__": | |
# import_data(".", "demo_products.csv", "demo_customer.csv", "demo_rental.csv") | |
app.run(port=8888) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import csv | |
from pprint import pprint | |
from pymongo import MongoClient | |
import threading | |
# https://stackoverflow.com/questions/35813854/how-to-join-multiple-collections-with-lookup-in-mongodb | |
""" | |
High Level Requirements | |
- Create a product database with attributes that reflect the contents of the csv file. | |
- Import all data in the csv files into your MongoDB implementation. | |
- Write queries to retrieve the product data. | |
- Write a query to integrate customer and product data. | |
Detail Tests | |
- a file called database.py | |
- includes functions like | |
import_data(directory_name, product_file, customer_file, rentals_file) | |
It returns 2 tuples: the first with a record count of the number of | |
products, customers and rentals added (in that order), the second with a count of any errors that occurred, in the same order. | |
""" | |
mongo = MongoClient("mongodb://localhost:27017") | |
db = mongo["norton"] | |
def import_data(data_dir, *files): | |
for filepath in files: | |
collection_name = filepath.split(".")[0] | |
print("opening", "/".join([data_dir, filepath])) | |
with open("/".join([data_dir, filepath])) as file: | |
reader = csv.reader(file, delimiter=",") | |
header = False | |
for row in reader: | |
if not header: | |
header = [h.strip("\ufeff") for h in row] | |
else: | |
data = {header[i]:v for i,v in enumerate(row)} | |
# print(data) | |
cursor = db[collection_name] | |
try: | |
cursor.insert_one(data) | |
except Exception as e: | |
print(e) | |
def import_data_multithreading(filepath): | |
collection_name = filepath.split(".")[0] | |
print("opening", filepath) | |
with open(filepath) as file: | |
reader = csv.reader(file, delimiter=",") | |
header = False | |
for row in reader: | |
if not header: | |
header = [h.strip("\ufeff") for h in row] | |
else: | |
data = {header[i]:v for i,v in enumerate(row)} | |
# print(data) | |
cursor = db[collection_name] | |
try: | |
cursor.insert_one(data) | |
except Exception as e: | |
print(e) | |
def get_product_info(product_id): | |
return db["product"].find_one({"product_id": product_id}) | |
def get_rental_info(): | |
return db["rental"].aggregate([ | |
{ | |
"$lookup": | |
{ | |
"from": "customer", | |
"localField": "user_id", # what is field name in rental? | |
"foreignField": "Id", # what is field name in customer? | |
"as": "customer_info" | |
} | |
}, | |
{ | |
"$lookup": | |
{ | |
"from": "product", | |
"localField": "product_id", # what is field name in rental? | |
"foreignField": "product_id", # what is field name in product? | |
"as": "product_info" | |
} | |
}, | |
]) | |
def show_available_products(): | |
# Returns a Python dictionary of products listed as available with the following fields: | |
# product_id. | |
# description. | |
# product_type. | |
# quantity_available. | |
# For example: | |
# {‘prd001’:{‘description’:‘60-inch TV stand’,’product_type’:’livingroom’,’quantity_available’:‘3’},’prd002’:{‘description’:’L-shaped sofa’,’product_type’:’livingroom’,’quantity_available’:‘1’}} | |
output = {} | |
for product in db["product"].find(): | |
output[product["product_id"]] = { | |
"description": product["description"], | |
"product_type": product["product_type"], | |
"qantity_available": product["qantity_available"], #### MISSPELLING!!!! | |
} | |
return output | |
def show_rentals(product_id): | |
# Returns a Python dictionary with the following user information from users that have rented products matching product_id: | |
# user_id. | |
# name. | |
# address. | |
# phone_number. | |
# email. | |
# For example: | |
# {‘user001’:{‘name’:’Elisa Miles’,’address’:‘4490 Union Street’,’phone_number’:‘206-922-0882’,’email’:’elisa.miles@yahoo.com’},’user002’:{‘name’:’Maya Data’,’address’:‘4936 Elliot Avenue’,’phone_number’:‘206-777-1927’,’email’:’mdata@uw.edu’}} | |
rentals = db["rental"].aggregate([ | |
{ | |
"$lookup": | |
{ | |
"from": "customer", | |
"localField": "user_id", # what is field name in rental? | |
"foreignField": "Id", # what is field name in customer? | |
"as": "customer_info" | |
} | |
}, | |
{ | |
"$match":{ | |
"$and":[{"product_id" : product_id}] | |
} | |
}, | |
]) | |
output = {} | |
for rental in rentals: | |
# DEBUG | |
pprint(rental["customer_info"]) | |
user_id = rental["customer_info"][0]["Id"] | |
name = rental["customer_info"][0]["Name"] + " " + rental["customer_info"][0]["Last_name"] | |
address = rental["customer_info"][0]["Home_address"] | |
phone_number = rental["customer_info"][0]["Phone_number"] | |
email = rental["customer_info"][0]["Email_address"] | |
output[user_id] = { | |
"name": name, | |
"address": address, | |
"phone_number": phone_number, | |
"email": email, | |
} | |
return output | |
if __name__ == "__main__": | |
db["customer"].drop() | |
db["product"].drop() | |
db["rental"].drop() | |
import_data("data", "product.csv", "customer.csv", "rental.csv") | |
# pprint(get_product_info("P000013")) | |
# show_available_products() | |
# pprint(show_rentals("P000001")) | |
# for rental in get_rental_info(): | |
# pprint(rental) | |
# MULTITHREADING | |
def func(): | |
for i in range(5): | |
print("hello from thread %s" % threading.current_thread().name) | |
time.sleep(1) | |
files = ["data/product.csv", "data/customer.csv", "data/rental.csv"] | |
threads = [] | |
for filepath in files: | |
thread = threading.Thread(target=import_data_multithreading, args=(filepath,)) | |
thread.start() | |
threads.append(thread) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment