Python: Flask Resource

This tutorial helps setup flask restful api’s.

Install Python Packages:

Open cmd and navigate into your testApp folder and run the following commands.

pip install flask-RESTful && pip freeze > requirements.txt
__init__.py:

On the init of your application you will need to setup flask_restful. There are config options you could set for config.py. Look into it!

from flask_restful import Api
api = Api(app)

#Add api endpoints
#Get
api.add_resource(home.views.MyResource, '/home/getMyData/')

#Post
api.add_resource(home.views.MyResource, '/home/getMyData/', methods=['POST'])
Setup home views.py:

You need to import Resource in it’s most simplistic form. However if you want to deal with request parameters add in reqparse and inputs. Inputs give you access to boolean that way a boolean can be parsed into a python boolean easily.

from flask_restful import Resource, reqparse, inputs

You can now use get, post, etc. I will give you three examples below.

Get:

class MyResource(Resource):
	def get(self):
		return {}

Get /w Parameter:

class MyResource(Resource):
	def get(self, var):
		return {}

Get /w Parameter & Request Parameter:

class MyResource(Resource):
	def get(self, var):
        	parser = reqparse.RequestParser()
        	parser.add_argument('new_request_var', type=str, default='')

        	#If you want to have a boolean request parameter do the following.
        	parser.add_argument('new_request_var_bool', type=inputs.boolean, default=False)

        	args = parser.parse_args(strict=True)
        	new_request_var = args['new_request_var']
        	new_request_var_bool = args['new_request_var_bool']

		return {}

Post:

class MyResource(Resource):
	def post(self):
		return {}

Python: Run Module as Script

We can run a Python module as a script directly. In order to do that it checks that “__name__” is set to “__main__”. If it is then it executes the code in the if block. See below for example.

#imports here

def run():
	#run code here
	return

if __name__ == "__main__":
    run()

Flask: React Website

This whole tutorial describes in depth how to create a React website with Python. You must have Eclipse installed before you continue. If you have it already installed and configured you can continue on. Note that you should probably have HTML Editor and TypeScript IDE installed for Eclipse.

FYI: I am using Windows at the moment for this tutorial but you can substitute Ubuntu in if you so chose.

We will require Postgres 9.4, nodejs before you continue. You can get Nodejs from here. You can get Postgres 9.4 from here.

In this tutorial we use Flask. If you are not familiar with it go here. Flask is a lightweight Python web framework based on Werkzeug and Jinja 2.

Folder Structure:
  • You will need to create a folder called “testApp”.
  • Inside testApp create the following:
    • folder: app
      • file: models.py
      • file: __init__.py
      • folder: auth
        • file: __init__.py
        • file: views.py
      • folder: home
        • file: __init__.py
        • file: views.py
        • folder: js
          • file: home.jsx
      • folder: templates
        • file: base.html
        • file: login.html
        • file: register.html
    • file: config.py
    • file: requirements.txt
    • file: webpack.config.js
    • file: run.py
    • folder: instance
      • file: config.py
    • folder: static
      • folder: common
        • folder: css
          • file: base.css
      • file: manifest.json
    • .babelrc
Install Python Packages:

Open cmd/terminal and navigate into your testApp folder and run the following commands.

pip install flask-sqlalchemy && pip freeze > requirements.txt
pip install flask-login && pip freeze > requirements.txt
pip install flask-migrate && pip freeze > requirements.txt
pip install psycopg2 && pip freeze > requirements.txt
pip install flask-Webpack && pip freeze > requirements.txt
pip install Flask-WTF && pip freeze > requirements.txt
pip install flask-bootstrap && pip freeze > requirements.txt

Couple things to note:

  • The “&& pip freeze > requirements.txt” saves the install in requirements.txt.
  • flask-migrate: database migrations package
  • flask-sqlalchemy: model engine for your database.
  • flask-login: provides user session management for Flask
  • psycopg2: Postgres driver
  • flask-Webpack: A Flask extension to manage assets with Webpack
  • flask-WTF: flask form validation
  • flask-bootstrap: An extension that includes Bootstrap in your project, without any boilerplate code
Routing:

For the purpose of this example we are using the basic Flask implementation. IE: @home.route(‘/’). However if you want to do a more advanced routing do Resource.

Database:

Open PgAdmin III and create yourself a database called “testApp”. Also create a user with password granting access to testApp database. Permission as you see fit. Don’t forget to write down your password :).

Setup Config.py:

Read here for SqlAlchemy configruation options. DEBUG is for flask debugging.

class Config(object):
    """
    This is the shared common configurations
    """

    # Put any configurations here that are common across all environments

class DevelopmentConfig(Config):
    """
    This is the development configurations
    """
    
    DEBUG = True
    SQLALCHEMY_ECHO = True #Display queries to console

class ProductionConfig(Config):
    """
    This is the production configurations
    """
    DEBUG = False
    SQLALCHEMY_ECHO = False #Do not Display queries to console

app_config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig
}

Things to Note:

Notice how we don’t have any database connections in this config.py. That is because we really shouldn’t checkin to source control our database connection strings, etc.

Setup instance config.py:

We open the file config.py inside our “instance” folder and add this line only.

SQLALCHEMY_DATABASE_URI = "postgresql://##USER##:##PASSWORD##@##HOST##/testApp"
Setup __init__.py in “app” folder:
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from flask_migrate import Migrate
from flask_bootstrap import Bootstrap
import os

#Importing from the config.py
from config import app_config

# db variable initialization
db = SQLAlchemy()

#The flask login manager
login_manager = LoginManager()

webpack = Webpack()

def create_app(config_name):
    #This will be either "development" or "production" mapped to what we write in the config.py application
    #static_folder is where the static folder will be located
    app = Flask(__name__, instance_relative_config=True, static_folder=os.path.join(os.getcwd(), "static"))
    print('Running in %s' % (config_name))
    app.config.from_object(app_config[config_name])
    app.config.from_pyfile('config.py')
    #You need a secret key to be able to utilise the database connection
    app.secret_key = 'Awesome App'
    Bootstrap(app)
    db.init_app(app)
    #This will make it so our chunked js files are able to be loaded on the template
    app.config.update({'WEBPACK_MANIFEST_PATH': '../manifest.json'})
    webpack.init_app(app)

    #if a user tries to access a page that they are not authorized to, it will redirect to the specified view and display the specified message.
    login_manager.init_app(app)
    login_manager.login_message = "You must be logged in to access this page."
    #auth.login is not the route but actually the class path.
    login_manager.login_view = "auth.login"
    
    #This let's us do our migrations
    migrate = Migrate(app, db)

    #Bring in our new tables
    from app import models

    #Our blueprints for our app

    #This is how you get authenticated
    from .auth import auth as auth_blueprint
    app.register_blueprint(auth_blueprint)
	
    #Bring in the home module
    from .home import home as home_blueprint
    app.register_blueprint(home_blueprint)

    return app
Environment Variable Setup:

Flask has two environment variables that you can set which point to the environment to use and the run.py file. These are FLASK_CONFIG and FLASK_APP. I set my environment to “development” and the app to use “run.py”. Notice that “development” maps to the key value pair in config.py.

Setup run.py:

Notice how we utilise the FLASK_CONFIG from the environment variables to setup our environment and grab the right config class.

import os #We need this to get the OS ENV VARIABLE 'FLASK_CONFIG'

#You are going to import the create_app from the __init__.py file
from app import create_app

#In our environment variables we create "FLASK_CONFIG" and set our value either development or production
config_name = os.getenv('FLASK_CONFIG')
app = create_app(config_name)

if __name__ == '__main__':
    app.run()

Now the fun really starts!!!

Setup models.py:

We setup our User model ensuring security of our password. Later on I will show you what happens with flask-migrate.

from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash

from app import db

class User(UserMixin, db.Model):
    """
    Create an Users table
    """

    # Ensures table will be named in plural and not in singular
    # as is the name of the model
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(60), index=True, unique=True)
    username = db.Column(db.String(60), index=True, unique=True)
    first_name = db.Column(db.String(60), index=True)
    last_name = db.Column(db.String(60), index=True)
    password_hash = db.Column(db.String(128))

    @property
    def password(self):
        """
        Prevent pasword from being accessed
        """
        raise AttributeError('password is not a readable attribute.')

    @password.setter
    def password(self, password):
        """
        Set password to a hashed password
        """
        self.password_hash = generate_password_hash(password)

    def verify_password(self, password):
        """
        Check if hashed password matches actual password
        """
        return check_password_hash(self.password_hash, password)

    def __repr__(self):
        return ''.format(self.username)
Generate User Table:

Using flask-migrate we can now auto generate our User table into Postgres testApp database.

  1. Run “flask db init” to create our migration repository
    1. A “migrations” folder is created automatically.
  2. Run “flask db migrate”
    1. This generates the migration script
  3. Run “flask db upgrade”
    1. This creates the missing migrations into the database. AKA the users table.
    2. You will notice another table also got created “alembic_version”. This is how it stores the version it is at.
  4. Confirm that the db was migrated successfully.
Create Our Templates:

We use the base.html file for our react pages using the “app” id on the div. The login.html is for our login form and register.html is to register a new user. You can use the base.css file in the static/common/css folder to style it how you want. I recommend using flex.

base.html:

<!DOCTYPE html>
{% import "bootstrap/utils.html" as utils %}
{% import "bootstrap/wtf.html" as wtf %}
{% extends "bootstrap/base.html" %}
<html>
	<head>
		{% block head %}
		{{ super() }}
		<title>{{title}}</title>
		<link rel="stylesheet" type="text/css" href={{ url_for("static", filename="common/css/base.css") }} />
		{% endblock %}
	</head>
	{% block content %}
	<div class="container-app">
		<div class="container-header">Team Link</div>
	    <div class="container-body" id="app"></div>
	    <script type="text/javascript" src={{ asset_url_for(script_name) }}></script>
	</div>
	{% endblock %}
</html>

login.html:

<!DOCTYPE html>
{% import "bootstrap/utils.html" as utils %}
{% import "bootstrap/wtf.html" as wtf %}
{% extends "bootstrap/base.html" %}
<html>
	<head>
		{% block head %}
		{{ super() }}
		<title>{{title}}</title>
		<link rel="stylesheet" type="text/css" href={{ url_for("static", filename="common/css/base.css") }} />
		{% endblock %}
	</head>
	{% block content %}
	<div class="container-app">
		<div class="container-header">My Awesome App</div>
	    <div class="container-body" id="app">
			<div class="panel panel-default">
				<div class="panel-heading">
					<h3 class="panel-title">Login</h3>
				</div>
				<div class="panel-body">
					{{ wtf.quick_form(form) }}
				</div>	    
				{{ utils.flashed_messages() }}
				Click here to <a href="/register">register</a>
			</div>
		</div>
	</div>
	{% endblock %}
</html>

register.html

<!DOCTYPE html>
{% import "bootstrap/utils.html" as utils %}
{% import "bootstrap/wtf.html" as wtf %}
{% extends "bootstrap/base.html" %}
<html>
	<head>
		{% block head %}
		{{ super() }}
		<title>{{title}}</title>
		<link rel="stylesheet" type="text/css" href={{ url_for("static", filename="common/css/base.css") }} />
		{% endblock %}
	</head>
	{% block content %}
	<div class="container-app">
		<div class="container-header">Team Link</div>
	    <div class="container-body" id="app">
			<div class="panel panel-default">
				<div class="panel-heading">
					<h3 class="panel-title">Register</h3>
				</div>
				<div class="panel-body">
    				{{ wtf.quick_form(form) }}
				</div>	    
				{{ utils.flashed_messages() }}
				Click here to <a href="login">login</a>
			</div>
		</div>
	</div>
	{% endblock %}
</html>
Setup home __init__.py:

This creates the blueprint that we have in app.__init__.py.

from flask import Blueprint

home = Blueprint('home', __name__)

#This is the views.py from the home directory.
from . import views
Setup home views.py:

@login_required let’s flask know that you need to be logged in to get to this page. Don’t forget to see “render_template” method. How it has “script_name” in it and it uses base.html template. “script_name” was utilised in base.html. It brings in our js file for us on each page we go to.

from flask import render_template
from flask_login import login_required

#This is our blueprint we setup in __init__.py
from . import home

@home.route('/')
@login_required
def homepage():
    """
    Render the homepage template on the / route
    """
    return render_template('base.html', script_name='home.js', title="Welcome")
Setup React home JSX file:

React uses jsx files. So in my home directory I have a js folder and inside that we have our home.jsx file. Let’s set that up to something really basic. Remember above I said in the “render_template” we use the div id “app”. The ReactDOM will put our class in that spot. I will show you later how that is done.

var React = require("react");
var ReactDOM = require("react-dom");
var createReactClass = require("create-react-class");

var Home = createReactClass({
	render: function() {
		return (<div>Hi</div>);
	}
});

ReactDOM.render(<Home />, document.getElementById("app"));
Node Init:

We need to go to our root directory testApp from command prompt and run the following in the exact order.

npm init
npm install react react-dom --save
npm install webpack webpack-dev-server --save
npm install --save-dev babel babel-core babel-loader babel-preset-es2015 babel-preset-react
npm install create-react-class --save
npm install bootstrap --save
npm install jquery --save
npm install clean-webpack-plugin --save-dev
npm install manifest-revision-webpack-plugin --save-dev
npm install sync-exec --save-dev
npm install uglifyjs-webpack-plugin --save-dev

Things to Note:

  • npm init: Creates package.json file
  • npm install –save: will save the install to package.json
  • Each package install went into a new directory called “node_modules”.
Setup webpack.config.js:

We want to transition our jsx files to .js. Remember back in app.__init__.py we setup the static folder location. Checkout the “path” key below and now we know how it knows where it’s assets are going to be located. Our entry key value pair is the locations of each of our jsx files to create assets from. Then we have our loaders.

var path = require("path");
var webpack = require('webpack');
var ManifestRevisionPlugin = require('manifest-revision-webpack-plugin');
const CleanWebpackPlugin = require('clean-webpack-plugin');

module.exports = {
	entry: {
		"home": [path.resolve(__dirname, "./app/home/js/home.jsx")],
	},
	output: {
        path: path.join(__dirname, "static"),
		publicPath: "/static/",
		filename: "[name]-[hash].js"
	},
	plugins: [
                new CleanWebpackPlugin(["static/*.js", "static/manifest.json"], {root: __dirname, verbose: true, dry: false, exclude: ["base.css"]}),
		new ManifestRevisionPlugin(path.resolve(__dirname, "./manifest.json"), {rootAssetPath: './static', ignorePaths: ['./node_modules']}),
		new webpack.NoEmitOnErrorsPlugin(),
		new UglifyJsPlugin(),
		new webpack.optimize.AggressiveMergingPlugin(),
		new webpack.HotModuleReplacementPlugin()
	],
	module: {
		loaders: [
			{
				test: /\.jsx?$/,
				exclude: /(node_modules)/,
				loader: 'babel-loader',
				query: {
					presets: ['react','es2015']
				}
			}
		]
	}
};
.babelrc

Set the following in your file at the root directory.

{
 "presets": [ "es2015", "react" ]
}
Let’s Test Node & Webpack:

Open command prompt and navigate to our testApp folder and run “webpack”. You will notice that a “static” folder is created in our root directory. In it we will now see “home.js” file. Remember back above we set __init__.py static folder and in home.views.py file we said in render_template script_name “home.js”. This is how it all maps together.

Setup auth __init__.py:

This creates the blueprint that we have in app.__init__.py.

from flask import Blueprint

auth = Blueprint('auth', __name__)

#This is the views.py from the auth directory.
from . import views

Setup auth views.py:

from flask import flash, redirect, render_template, url_for
from flask_login import login_required, login_user, logout_user
from flask_wtf import FlaskForm
from wtforms import PasswordField, StringField, SubmitField, ValidationError
from wtforms.validators import DataRequired, Email, EqualTo
from .. import db, login_manager
from ..models import User

from . import auth

class RegistrationForm(FlaskForm):
    """
    Form for users to create new account
    """
    email = StringField('Email', validators=[DataRequired(), Email()])
    username = StringField('Username', validators=[DataRequired()])
    first_name = StringField('First Name', validators=[DataRequired()])
    last_name = StringField('Last Name', validators=[DataRequired()])
    password = PasswordField('Password', validators=[
                                        DataRequired(),
                                        EqualTo('confirm_password')
                                        ])
    confirm_password = PasswordField('Confirm Password')
    submit = SubmitField('Register')

    def validate_email(self, field):
        if User.query.filter_by(email=field.data).first():
            raise ValidationError('Email is already in use.')

    def validate_username(self, field):
        if User.query.filter_by(username=field.data).first():
            raise ValidationError('Username is already in use.')

class LoginForm(FlaskForm):
    """
    Form for users to login
    """
    email = StringField('Email', validators=[DataRequired(), Email()])
    password = PasswordField('Password', validators=[DataRequired()])
    submit = SubmitField('Login')

@login_manager.user_loader
def load_user(id):
    #This is the how we locate the user in our testApp database
    return User.query.get(int(id))

@auth.route('/register', methods=['GET', 'POST'])
def register():
    form = RegistrationForm()
    if form.validate_on_submit():
        user = User(email=form.email.data,
                            username=form.username.data,
                            first_name=form.first_name.data,
                            last_name=form.last_name.data,
                            password=form.password.data)

        # add user to the database
        db.session.add(user)
        db.session.commit()
        flash('You have successfully registered! You may now login.')

        # redirect to the login page
        return redirect(url_for('auth.login'))

    # load registration template
    return render_template('register.html', form=form, title='Register')

@auth.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()
    if form.validate_on_submit():
        # check whether user exists in the database and whether
        # the password entered matches the password in the database
        user = User.query.filter_by(email=form.email.data).first()
        if user is not None and user.verify_password(
                form.password.data):
            # log user in
            login_user(user)

            # redirect to the dashboard page after login
            return redirect("/")

        # when login details are incorrect
        else:
            flash('Invalid email or password.')

    # load login template
    return render_template('login.html', form=form, title='Login')

@auth.route('/logout')
@login_required
def logout():
    """
    Handle requests to the /logout route
    Log an user out through the logout link
    """
    logout_user()
    flash('You have successfully been logged out.')

    # redirect to the login page
    return redirect(url_for('auth.login'))
Let’s Run our App:

Open command prompt navigate to our testApp folder and run “flask run”. If no mistakes were made you should now be able to navigate to our site. http://localhost:5000.

 

Resources:

In learning how to incorporate flask into python I used the following part one only tutorial as a guide. Very helpful.

Python IDE Installation for Eclipse

This tutorial will guide you through configuring Eclipse for Python. Ensure you have followed the tutorial on installing Eclipse first.

You need to install PyDev. Open Eclipse. Click Help–>Install New Software. In the “work with” put “http://pydev.org/updates” and click Add. Follow the prompts and you are done for now.

You also need to ensure you have installed Python 3.6. You can get it from here. You should add the environment variable “PYTHON_HOME” and enter the python directory. I would also add “%PYTHON_HOME%\;%PYTHON_HOME%\Scripts” to your “path” environment variable.

I would also at this time install PIP. Just to make sure you have everything you will need. You will need to download “get-pip.py“. You will need to put it into the “Scripts” folder where you install Python 3.6. Navigate into your Scripts folder and then run “python get-pip.py”.

Once PyDev is installed and you have restarted Eclipse. Open PyDev perspective. Go to Window–>Perspective–>Open Perspective–>Other. Then select PyDev and click Ok.

Optional:

TypeScript IDE:

Used for React. It’s really handy. Open Eclipse. Click Help–>Install New Software. In the “work with” put “http://oss.opensagres.fr/typescript.ide/1.1.0/” and click Add. Follow the prompts and you are done for now. make sure you select “Embed Node.js” and “TypeScript IDE”.

HTML Editor:

Used for HTML files. Click Help –> Eclipse Marketplace. Search for “HTML Editor”. Then click install. After it is installed and Eclipse is restarted Click Window –> Preferences –> Web. Under “HTML Files” change encoding to “ISO 10646/Unicode(UTF-8). Under “Editor” add “div”. You can get more info and configuration from here.

Python: Misc

This post will just be for misc basic python stuff.

Let’s say you want to raise 10 to the power of 3.

10 ** 3

If you want to check if a variable is of a specific type. You can use “isinstance”:

#Check if it is int
isinstance(variable, int)
#Check if it is  string
isinstance(variable, basestring)

If you want to test that a dictionary is not empty

bool(my_dict)

Check if a file exists

import os
os.path.isfile(path)

Recursively loop through a directory to get the files with path to a list

import os

x = [os.path.join(r,file) for r,d,f in os.walk("E:") for file in f]

for val in x:
    print(val)

Python: Connect To Hadoop

We can connect to Hadoop from Python using PyWebhdfs package. For the purposes of this post we will use version 0.4.1. You can see all API’s from here.

To build a connection to Hadoop you first need to import it.

from pywebhdfs.webhdfs import PyWebHdfsClient

Then you build the connection like this.

HDFS_CONNECTION = PyWebHdfsClient(host=##HOST## port='50070', user_name=##USER##)

To list the contents of a directory you do this.

HDFS_CONNECTION.list_dir(##HADOOP_DIR##)

To pull a single file down from Hadoop is straight forward. Notice how we have the “FileNotFound” brought in. That is important when pulling a file in. You don’t actually need it but “read_file” will raise that exception if it is not found. By default we should always include this.

from pywebhdfs.errors import FileNotFound

try:
	file_data = HDFS_CONNECTION.read_file(##FILENAME##)
except FileNotFound as e:
	print(e)
except Exception as e:
	print(e)

 

 

AWS: Python S3

This entry is part 3 of 3 in the series AWS & Python

If you haven’t already done so please refer to the AWS setup section which is part of this series. As time goes on I will continually update this section.

To work with S3 you need to utilise the “connection_s3” you setup already in the previous tutorial on setting up the connection.

To load a file from a S3 bucket you need to know the bucket name and the file name.

connection_s3.Object(##S3_BUCKET##, ##FILE_NAME##).load()

If you want to check if the check if a file exists on S3 you do something like the below. However you will need to import botocore.

import botocore

def keyExists(key):
    file = connection_s3.Object(##S3_BUCKET##, ##FILE_NAME##)
    
    try:
        file.load()
    except botocore.exceptions.ClientError as e:
        exists = False
    else:
        exists = True
    
    return exists, file

If you want to copy a file from one bucket to another or sub folder you can do it like below.

connection_s3.Object(##S3_DESTINATION_BUCKET##, ##FILE_NAME##).copy_from(CopySource=##S3_SOURCE_BUCKET## + '/' + ##FILE_NAME##)

If you want to delete the file you can use the “keyExists” function above and then just call “delete”.

##FILE##.delete()

If you want to just get a bucket object. Just need to specify what bucket and utilise the S3 connection.

bucket = connection_s3.Bucket(##S3_BUCKET##)

To upload a file to S3’s bucket. You need to set the body, type and name. Take a look at the below example.

bucket.put_object(Body=##DATA##,ContentType="application/zip", Key=##FILE_NAME##)

If you want to loop over the objects in a bucket. It’s pretty straight forward.

for key in bucket.objects.all():
	file_name = key.key
	response = key.get()
	data = response_get['Body'].read()

If you want to filter objects in a bucket.

for key in bucket.objects.filter(Prefix='##PREFIX##').all():
        file_name = key.key
	response = key.get()
	data = response_get['Body'].read()

AWS: Python Kinesis Streams

This entry is part 2 of 3 in the series AWS & Python

If you haven’t already done so please refer to the AWS setup section which is part of this series. As time goes on I will continually update this section.

To put something on the Kinesis Stream you need to utilise the “connection_kinesis” you setup already in the previous tutorial on setting up the connection. You will need to set the partition key, data and stream.

response = connection_kinesis.put_record(StreamName=##KINESIS_STREAM##, Data=##DATA##, PartitionKey=##FILE_NAME##)

Depending on your data you may need to utf8 encode. For example below.

bytearray(##MY_DATA##, 'utf8')

To read from the kinesis stream you need to setup the shard iterator then retrieve the data from the stream. Not forgetting to grab the new shard iterator from the returned records. Remember to not query against the queue to fast.

#shardId-000000000000 is the format of the stream
shard_it = connection_kinesis.get_shard_iterator(StreamName=##KINESIS_STREAM##, ShardId='shardId-000000000000', ShardIteratorType='LATEST')["ShardIterator"]

recs = connection_kinesis.get_records(ShardIterator=shard_it, Limit=1)

#This is the new shard iterator returned after queueing the data from the stream.
shard_it = out["NextShardIterator"]

 

AWS: Python Setup

This entry is part 1 of 3 in the series AWS & Python

When you want to work with S3 or a Kinesis Stream we first need to setup the connection. At the time of this writing I am using boto3 version 1.3.1.

Next we need to import the package.

import boto3

Next we setup the session and specify what profile we will be using.

profile = boto3.session.Session(profile_name='prod')

The profile name comes from the “credentials” file. You can set the environment variable “AWS_SHARED_CREDENTIALS_FILE” to specify what credentials file to use. You can setup the credentials file like below. You can change the “local” to anything you want. I normally use “stage”, “dev” or “prod”.

[local]
aws_access_key_id=##KEY_ID##
aws_secret_access_key=##SECRET_ACCESS_KEY##
region=##REGION##

Next we need to setup the connection to S3. To do this we will need to use the profile we created above.

connection_s3 = profile.resource('s3')

If we want to also use a Kinesis stream then we need to setup the connection. To do this we will need the profile we created above.

connection_kinesis = profile.client('kinesis')

Python: Enums

There are a variety of different ways to create enums. I show you a few different ways. They are not full version but you get the idea. If you have another way please feel free to add.

Option 1:

def enum(**enums):
    return type('Enum', (), enums)

my_enum = enum(NONE=0, SOMEVAL=1, SOMEOTHERVAL=2)

Option 2:
You will notice that you pass in “Enum” into the class. Also this way you can also declare classmethods for getting enum from string or tostring. It’s really your choice how you get the string representation you could either use str(MyEnum.VALUE) or MyEnum.tostring()

from enum import Enum

class MyEnum(Enum):
    VALUE = 0

    def __str__(self):
        if self.value == MyEnum.VALUE:
            return 'Value'
        else:
            return 'Unknown ({})'.format(self.value)
    def __eq__(self,y):
        return self.value==y

    @classmethod
    def fromstring(cls, value):
        """
        Converts string to enum
        """

        return getattr(cls, value.upper(), None)

    @classmethod
    def tostring(cls, val):
        """
        Converts enum to string
        """

        for k, v in vars(cls).iteritems():
            if v == val:
                return k

Option 3:

class MyEnum():
    NONE = 1
    VALUE = 2
    
    def __init__(self, Type):
        if Type is None:
            self.value = MyEnum.VALUE
        else:
            self.value = int(Type)
    def __str__(self):
        if self.value == MyEnum.VALUE:
            return 'Value'
        else:
            return 'None'
    def __eq__(self,y):
        return self.value==y

Python: For Loops

There are a variety of different ways to write a for loop in python. See below for different options. If you have any other suggestions please feel to add a comment.

Loop through a list one record at a time.

for rec in data:
	#Do Something

Loop through a range of numbers up to 100.

for i in range(100)
	#Do Something

Enumerate a list getting the index and value.

for idx, value in enumerate(data):
	#Do Something

Inline for loop.

d = [ x for x in data]

Inline for loop with an if condition.

d = [ x for x in data if x==1]

Python: Flask SQLAlchemy

If you are using SQLAlchemy then no doubt you’ve ran into some questions. Find below some more common usages.

Installation:

pip install flask-sqlalchemy && pip freeze > requirements.txt

Table Definition:
You will notice that we are creating a unique constraint, primary keys, foreign keys.

from sqlalchemy.orm import relationship
from sqlalchemy.dialects.postgresql import BOOLEAN, BIT, DATE, JSON, DOUBLE_PRECISION, UUID
from sqlalchemy.sql.functions import current_timestamp
from sqlalchemy.schema import UniqueConstraint

class MyTable(db.Model):
    __tablename__ = 'mytable'
    __table_args__ =  (UniqueConstraint('column_name'),)
    
    primarykey_id = db.Column(db.Integer, primary_key=True, autoincrement="True")
    bit_column = db.Column(BIT)
    text_column = db.Column(db.Text)
    foreignkey_id = db.Column(db.Integer, db.ForeignKey('table.column'))
    foreignkey = relationship("table", foreign_keys=[foreignkey_id ])
    bool_column = db.Column(BOOLEAN)
    created_date = db.Column(db.DateTime, default=current_timestamp())
    guid_column = db.Column(UUID)
    bigint_column = db.Column(db.BigInteger)
    double_precision_column = db.Column(DOUBLE_PRECISION)
    json_column = db.Column(JSON)
    string_column = db.Column(db.String(200))
    
    def __init__(self, primarykey_id, bit_column):
        self.primarykey_id = primarykey_id
        self.bit_column = bit_column

    def __repr__(self):
        return '<MyTable primarykey_id %r Name %r bit_column %r>' % (self.primarykey_id, self.bit_column)

Database Connections:
The default connection is the “SQLALCHEMY_DATABASE_URI”. However you can attach to other DB by using the BINDS as demonstrated below.

SQLALCHEMY_DATABASE_URI = "postgresql://##USER##:##PASS##@##SERVER##/##DB##"

SQLALCHEMY_BINDS = {
	"NAME": "postgresql://##USER##:##PASS##@##SERVER##/##DB##",
}

b.session().execute("", params=dict(), bind=db.get_engine(app, 'NAME'))

Concatenation:

 from sqlalchemy.sql.expression import func

func.concat(Table.Column,'-',Table.Column2)

Case Statements:

 from sqlalchemy.sql.expression import case

case([(column1 == 'some_val', 1)], else_=0)

Ordering:
The example below does multiple columns but you can do just one if need be.

 order_by(Table.Column1.desc(), Table.Column2.asc())

Literals:
If you want to add NULL as a column output you must use literals. You could do this with a numerical data or string or whatever.

 from sqlalchemy import literal

literal(None).label('column_name')

Retrieving Records:

 .first()
.one()
.fetchone()

.all()
.fetchall()

Joins:

 join(Table2, and_(Table2.column_id==Table1.column_id))
outerjoin(Table2, and_(Table2.column_id==Table1.column_id))

Union:

 query1.union(query2)

Select Fields:
This just selects one column but you can comma separate it.

 .with_entities(Table.Column)

Where:
You don’t have to use and_ or or_. Use them as needed.

 from sqlalchemy.sql.expression import and_, or_

.filter(and_(Table.Column1, Table.Column2, or_(Table.Column3_, Table.Column4))))

Not:
Select data where value is not NULL

 Table.Column.isnot(None)

Coalesce:

 from sqlalchemy.sql.functions import  coalesce

coalesce(Table.Column, 'Unknown')

Sum:

 from sqlalchemy.sql.expression import func

func.sum(Table.Column)

Count:

 from sqlalchemy.sql.expression import func

func.count()
func.count('*')

Max:

 from sqlalchemy.sql.expression import func

func.max(Table.Column)

AVG:

 from sqlalchemy.sql.expression import func

func.avg(Table.Column)

Substring:

 from sqlalchemy.sql.expression import func

func.substring(Table.Column, REGEX)

Like:

 Table.Column.like("value")

Not Like:

 from sqlalchemy.sql.expression import not_

not_(Table.Column.like("value"))

Commit & Rollback:

 #commit
session.commit()

#rollback
session.rollback()

Inline SQL:

 .execute('SELECT * FROM SP(:param)', params=dict(param=1))

Cast:

 from sqlalchemy.sql.functions import Cast

Cast(Table.Column, BIT)

Array with Distinct:

 from sqlalchemy.sql.expression import distinct, func

func.array_agg(distinct(Table.Column)).label('column_name')

Python: Create in Memory Zip File

Sometimes we need to create a zip file in memory and save to disk or send to AWS or wherever we need to save it. Below is basic steps to do this.

from zipfile import ZipFile
from io import BytesIO

in_memory = BytesIO()
zf = ZipFile(in_memory, mode="w")

#If you have data in text format that you want to save into the zip as a file
zf.writestr("name", file_data)

#Close the zip file
zf.close()

#Go to beginning
in_memory.seek(0)

#read the data
data = in_memory.read()

#You can save it to disk
with open('file_name.zip','wb') as out:
      out.write(data)

Java: Embed Python

Let’s say you want to embed Python code in your application. You will use Jython.

pom.xml:

<dependency>
      <groupId>org.python</groupId>
      <artifactId>jython-standalone</artifactId>
      <version>2.7.0</version>
</dependency>

*.Java

import org.python.util.PythonInterpreter;
import org.python.core.*;

private static PythonInterpreter interpreter = new PythonInterpreter();

interpreter = new PythonInterpreter(null, new PySystemState());

//You put the key to register in JavaScript and pass the variable in
interpreter.set(KEY, VARIABLE);

//If you wanted to use this in the mapper as an example you would pass the key, value and context to the JavaScript function. That way when you write to the context in Python it writes it to the applications context.

interpreter.exec(PYTHONCODE);