{"id":274,"date":"2021-03-20T19:33:09","date_gmt":"2021-03-20T19:33:09","guid":{"rendered":"https:\/\/andrejacobs.org\/?p=274"},"modified":"2022-04-11T20:24:23","modified_gmt":"2022-04-11T20:24:23","slug":"100-days-of-learning-day-12-using-sqlalchemy-as-an-orm-for-our-lego-database","status":"publish","type":"post","link":"https:\/\/andrejacobs.org\/100-days-challenge\/100-days-of-learning-day-12-using-sqlalchemy-as-an-orm-for-our-lego-database\/","title":{"rendered":"100 Days of Learning: Day 12 \u2013 Using SQLAlchemy as an ORM for our Lego database"},"content":{"rendered":"\n

Here is my Log book<\/a><\/p>\n

Git repository<\/a> for the functions.<\/p>\n

Remember:<\/strong> faas-cli logs<\/code> is your best friend. You can also tail the logs with "faas-cli logs NAME \u2014tail"<\/p>\n

Using SQLAlchemy<\/h2>\n

First I need to read the tutorial<\/a> on how this actually works. I know SQLAlchemy is very popular but for some reason I have just never used it. I will also be referencing the docs on mariadb<\/a>.<\/p>\n

I also found this tutorial<\/a> good enough to learn the basic SQLAlchemy ORM that I needed for this project.<\/p>\n

# First trying this on my local machine\n(venv)$ pip install sqlalchemy\n(venv)$ python\n\n>>> from sqlalchemy import create_engine\n>>> from sqlalchemy import text\n>>> engine = create_engine("mariadb+mariadbconnector:\/\/user:pass@some_mariadb\/dbname?charset=utf8mb4")\n>>> with engine.connect() as conn:\n...     result = conn.execute(text("select 'hello world'"))\n...     print(result.all())\n<\/code><\/pre>\n
Ok that worked. I continued to explore what the most minimal ORM code I would need to write to have a model named LegoSet that will map to the table in MariaDB. I will spare you all the steps I took to establish this.<\/p>\n
I created a shell script named up.sh<\/code> to do the faas-cli up<\/code> with the build options required.<\/p>\n
#!\/bin\/bash\nfaas-cli up --build-option dev -f legodb.yml\n<\/code><\/pre>\n
Don’t forget to update the OpenFaaS requirements.txt file.<\/p>\n
(venv)$ pip freeze > legodb\/requirements.txt\n(venv)$ cat legodb\/requirements.txt\ngreenlet==1.0.0\nmariadb==1.0.6\nSQLAlchemy==1.4.2\n<\/code><\/pre>\n
Creating the LegoSet model class<\/h2>\n
Create a new file named models.py<\/code> in the same directory as handler.py<\/code>.<\/p>\n
from sqlalchemy import create_engine, Column, Integer, String\nfrom sqlalchemy.ext.declarative import declarative_base\nfrom sqlalchemy.orm import sessionmaker\n\nBase = declarative_base()\n<\/code><\/pre>\n
We will need to be able to import this file into handler.py. There is a good example on OpenFaaS<\/a> of how relative imports work in Python.<\/p>\n
First I tested that the local import and SQLAlchemy imports worked before moving on.<\/p>\n
# Modified handler.py\n...\nimport mariadb\n# Added this line to import the LegoSet model class\nfrom .models import LegoSet\n\n<\/code><\/pre>\n
Build and test<\/p>\n
$ .\/up.sh\n$ curl -s http:\/\/192.168.64.4:9999\/function\/legodb\/legosets | jq\n# Still works (it did take some trial and error to get here)\n<\/code><\/pre>\n
Getting all the Lego sets from the database<\/h2>\n
Edit models.py<\/code><\/p>\n
from sqlalchemy import create_engine, Column, Integer, String\nfrom sqlalchemy.ext.declarative import declarative_base\nfrom sqlalchemy.orm import sessionmaker\n\nBase = declarative_base()\n\nclass LegoSet(Base):\n    __tablename__ = 'legosets'\n    id = Column('pkID', Integer, primary_key=True, nullable=False)\n    legoID = Column('LegoID', Integer, unique=True, nullable=False)\n    description = Column('Description',String(200))\n    productURL = Column('ProductURL',String(4096))\n    imageURL = Column('ImageURL',String(4096))\n\ndef database_session(host, port, user, password, schema):\n    engine = create_engine(f"mariadb+mariadbconnector:\/\/{user}:{password}@{host}:{port}\/{schema}?charset=utf8mb4")\n    \n    Session = sessionmaker(bind=engine)\n    Base.metadata.create_all(engine)\n    \n    session = Session()\n    return session\n\ndef get_all_legosets(session):\n    legosets = session.query(LegoSet).all()\n    return legosets\n<\/code><\/pre>\n
Edit handler.py<\/code><\/p>\n
import os\nimport json\nfrom .models import LegoSet, database_session, get_all_legosets\n\n# GET \/legosets : Returns the list of lego sets\n# POST \/legoset : Add a new lego set to the database\n\ndef handle(event, context):\n    response = {'statusCode': 400}\n\n    if event.method == 'GET':\n        if event.path == '\/legosets':\n            legosets = get_list_of_legosets()\n            response = {'statusCode': 200, \n                'body': legosets,\n                'headers': {'Content-Type': 'application\/json'}\n                }\n    elif event.method == 'POST':\n        if event.path == '\/legoset':\n            response = add_new_legoset(event.body)\n    \n    return response\n\ndef load_secret(name):\n    filepath = os.path.join('\/var\/openfaas\/secrets\/', name)\n    with open(filepath) as f:\n        secret = f.read()\n    return secret\n\ndef create_database_session():\n    host = load_secret('database-host')\n    user = load_secret('database-user')\n    password = load_secret('database-password')\n    schema = os.environ.get('database-name')\n    port = int(os.environ.get('database-port', '3306'))\n\n    session = database_session(host, port, user, password, schema)\n    return session\n\ndef get_list_of_legosets():\n    session = create_database_session()\n    legosets = get_all_legosets(session)\n    result = []\n\n    for legoset in legosets:\n        result.append({\n            'legoID': legoset.legoID,\n            'description': legoset.description,\n            'productURL': legoset.productURL,\n            'imageURL': legoset.imageURL\n        })\n\n    session.close()\n    return {"sets": result}\n\ndef add_new_legoset(body):\n    response = None\n    try:\n        inputJSON = json.loads(body.decode('utf8').replace("'", '"'))\n        response = {\n            'statusCode': 200,\n            'body': {'received': inputJSON},\n            'headers': {'Content-Type': 'application\/json'}\n        }\n    except ValueError:\n        response = {\n            'statusCode': 400,\n            'body': {'reason': 'Invalid JSON'},\n            'headers': {'Content-Type': 'application\/json'}\n         }\n    return response\n<\/code><\/pre>\n
To verify that this does actually work as expected, I added an extra Lego set using MySQLWorkbench.<\/p>\n
Build and test.<\/p>\n
$ curl -s http:\/\/192.168.64.4:9999\/function\/legodb\/legosets | jq\n{\n  "sets": [\n    {\n      "description": "Pirates of Barracuda Bay",\n      "imageURL": "https:\/\/live.staticflickr.com\/65535\/49698377257_be3f773feb_b.jpg",\n      "legoID": 21322,\n      "productURL": "https:\/\/www.lego.com\/en-gb\/product\/pirates-of-barracuda-bay-21322"\n    },\n    {\n      "description": "Medieval Blacksmith",\n      "imageURL": "https:\/\/images.brickset.com\/sets\/AdditionalImages\/21325-1\/21325-1_95850.jpg",\n      "legoID": 21325,\n      "productURL": "https:\/\/www.lego.com\/en-gb\/product\/medieval-blacksmith-21325"\n    },\n    {\n      "description": "Mars Research Shuttle",\n      "imageURL": "https:\/\/www.lego.com\/cdn\/cs\/set\/assets\/blt32796f487437e9db\/60226_alt1.jpg",\n      "legoID": 60226,\n      "productURL": "https:\/\/www.lego.com\/en-gb\/product\/mars-research-shuttle-60226"\n    }\n  ]\n}\n<\/code><\/pre>\n
\"\"<\/p>\n
Time to add new Lego sets using our model<\/h2>\n
Edit model.py<\/code><\/p>\n
# Add this at the bottom\ndef create_legoset(json):\n    legoID = json.get('legoID', None)\n    if legoID is None:\n        return None\n    description = json.get('description', None)\n    if description is None:\n        return None\n    productURL = json.get('productURL', None)\n    imageURL = json.get('imageURL', None)\n\n    legoset = LegoSet(legoID=legoID, description=description, productURL=productURL, imageURL=imageURL)\n    return legoset\n<\/code><\/pre>\n
Edit handler.py<\/code><\/p>\n
# Change the import\nfrom .models import LegoSet, database_session, get_all_legosets, create_legoset\n\n# Replace add_new_legoset\ndef add_new_legoset(body):\n    response = None\n    try:\n        inputJSON = json.loads(body.decode('utf8').replace("'", '"'))\n        legoset = create_legoset(inputJSON)\n        if legoset is None:\n            raise ValueError()\n        else:\n            session = create_database_session()\n            session.add(legoset)\n            session.commit()\n            session.refresh(legoset)\n            newID = legoset.id\n            session.close()\n\n            response = {\n                'statusCode': 200,\n                'body': {'pkID': newID},\n                'headers': {'Content-Type': 'application\/json'}\n        }\n    except ValueError:\n        response = {\n            'statusCode': 400,\n            'body': {'reason': 'Invalid JSON'},\n            'headers': {'Content-Type': 'application\/json'}\n         }\n    return response\n<\/code><\/pre>\n
Build and test<\/p>\n
# Add a new lego set to the database\n$ curl -s -H "Content-Type: application\/json" -d '{"legoID":6929,"description":"Star Fleet Voyager","productURL":"https:\/\/brickset.com\/sets\/6929-1\/Star-Fleet-Voyager","imageURL":"https:\/\/images.brickset.com\/sets\/images\/6929-1.jpg"}' "http:\/\/192.168.64.4:9999\/function\/legodb\/legoset" | jq\n{"pkID":8}\n\n# Verify\n$ curl -s http:\/\/192.168.64.4:9999\/function\/legodb\/legosets | jq\n...\n    {\n      "description": "Star Fleet Voyager",\n      "imageURL": "https:\/\/images.brickset.com\/sets\/images\/6929-1.jpg",\n      "legoID": 6929,\n      "productURL": "https:\/\/brickset.com\/sets\/6929-1\/Star-Fleet-Voyager"\n    }\n...\n\n<\/code><\/pre>\n
Next up<\/h2>\n
We now have a MariaDB instance running on AWS RDS and an OpenFaaS function that serves a RESTful API that allows us to get the list of Lego sets as well as being able to add a new set to the database.<\/p>\n
Somethings that we could add (but I will leave that for the reader):<\/p>\n