Jupyter – Finding the point when a line graph crosses the threshold

A friend of came up with the problem. There are a set of points [(x, y), (x1, y1), (x2, …]. He wanted to find the points at which this line would pass the value Z less than the peak. If the maximum value is 100 and Z = 20. He wanted to find the points where it would cross y = 80.

Now there are multiple ways to solve this problem. I attempted a simple linear interpolation solution.

I don’t the solution itself to be a big thing. What I am really impressed is, how neatly I was able to present the solution using Jupyter Notebook to him.

I was able to document the solution in a step by step fashion, with visual representation of how I solved it.

Take a look

from IPython.display import display
import matplotlib.pyplot as plt

f = [100, 102, 103.5, 105.5, 106.5, 107.5, 108.5, 110]
mag = [0, 30, 40, 145.3, 166.5, 164.5, 75.79, 65.3]

fig, ax = plt.subplots()
ax.plot(f, mag)

for x,y in zip(f, mag):
    label = ax.text(x, y, y)



gap = 30

# 1. Find the maximum value
max_mag = max(mag)

# 2. Set the threshold value
y = max_mag - gap

ax.hlines(y, f[0], f[-1], linewidth=0.5, color="cyan")


max_idx = mag.index(max_mag)

# 4. Find the left and right values which are lower than the "y" you are looking for
left_start_idx = None
left_end_idx = max_idx
right_start_idx = max_idx
right_end_idx = None

for i in range(max_idx):
    left_idx = max_idx - i
    right_idx = max_idx + i

    # if left index is more than Zero (array left most is 0) and left is not yet set
    if left_idx >= 0 and not left_start_idx:
        value = mag[left_idx]
        # if the value is lower than our threshold then pickup the point
        # and the one next to it
        # that will form our segment to interoploate
        if value < y:  
            left_start_idx = left_idx
            left_end_idx = left_idx + 1

    # if the right index is less than our array size (0..N) and right is not yet set
    if right_idx < len(mag) and not right_end_idx:
        value = mag[right_idx]
        if value < y:
            right_end_idx = right_idx
            right_start_idx = right_idx - 1

if not right_end_idx:
    print("Cannot find point on the right lower than %d" % (y))

if not left_start_idx:
    print("Cannot find point on the left lower than %d" % (y))

# Plotting the lines we will be interpolating

if left_mag and right_mag:
        [f[left_start_idx], f[left_end_idx]],
        [mag[left_start_idx], mag[left_end_idx]],
        [f[right_start_idx], f[right_end_idx]],
        [mag[right_start_idx], mag[right_end_idx]],



Now Let us use the line equation

\frac{y - y1}{x - x1} = \frac{y2 - y1}{x2 - x1}

Solving for x we get

x = x1 + (x2 - x1) \frac{y - y1}{y2 - y1}

# Left point interpolation

y1 = mag[left_start_idx]
y2 = mag[left_end_idx]
x1 = f[left_start_idx]
x2 = f[left_end_idx]

x = x1 + (x2 - x1) * (y - y1) / (y2 - y1)

ax.scatter([x], [y], color="green")


# Right point interpolation

y1 = mag[right_start_idx]
y2 = mag[right_end_idx]
x1 = f[right_start_idx]
x2 = f[right_end_idx]

x = x1 + (x2 - x1) * (y - y1) / (y2 - y1)

ax.scatter([x], [y], color="green")


I was able to export the whole thing as a PDF and send it to him.

Simplifying a Factory Pattern function that has grown complex

This is a combination of the problem that I posted in Dev.to and StackExchange and the final solution that I adopted.

The Problem

I have a function which takes the incoming request, parses the data and performs an action and posts the results to a webhook. This is running as background as a Celery Task. This function is a common interface for about a dozen Processors, so can be said to follow the Factory Pattern. Here is the psuedo code:

processors = {
    "action_1": ProcessorClass1, 
    "action_2": ProcessorClass2,

def run_task(action, input_file, *args, **kwargs):
    # Get the input file from a URL
    log = create_logitem()
        file = get_input_file(input_file)
        log.status = "Failure"

    # process the input file
        processor = processors[action](file)
        results = processor.execute()
        log.status = "Failure"

    # upload the results to another location
        log.status = "Failure"

    # Post the log about the entire process to a webhoook

This has been working well for most part as the the inputs were restricted to action and a single argument (input_file). As the software has grown, the processors have increased and the input arguments have started to vary. All the new arguments are passed as keyword arguments and the logic has become more like this.

    input_file = get_input_file(input_file)
    if action == "action_2":
       input_file_2 = get_input_file(kwargs.get("input_file_2"))
    log.status = "failure"

    processor = processors[action](file)
    if action == "action_1":
        extra_argument = kwargs.get("extra_argument")
        results = processor.execute(extra_argument)
    elif action == "action_2":
        extra_1 = kwargs.get("extra_1")
        extra_2 = kwargs.get("extra_2")
        results = processor.execute(input_file_2, extra_1, extra_2)
        results = processor.execute()
    log.status = "Failure"

Adding the if conditions for a couple of things didn’t make a difference, but now almost 6 of the 11 processors have extra inputs specific to them and the code is starting to look complex and I am not sure how to simplify it. Or if at all I should attempt at simplifying it.

Something I have considered:
1. Create a separate task for the processors with extra inputs – But this would mean, I will be repeating the file fetching, logging, result upload and webhook code in each task.
2. Moving the file download and argument parsing into the BaseProcessor – This is not possible as the processor is used in other contexts without the file download and webhooks as well.

The solution

I solved it by making two important changes:

  1. Normalised the processor’s by making the common arguments positional and everything else keyword based. This allows me to pass the kwargs as I receive them without unpacking. It is the processor’s job.
  2. For the extra files, make a copy of the kwargs and replace the remote file url with the local file location. This way, the extra files are a part of the kwargs dict itself.
def run_task(action, input_file, *args, **kwargs):

    params = kwargs.copy()

    # Get the input file from a URL
    log = create_logitem()
        file = get_input_file(input_file)
        if action == "action_2":
           params["extra_file"] = get_input_file(kwargs["extra_file"]  # update the files in params
        log.status = "Failure"

    # process the input file
        processor = processors[action](file)
        results = processor.execute(**params)   # Unpack and pass the params
        log.status = "Failure"

    # upload the results to another location
        log.status = "Failure"

    # Post the log about the entire process to a webhoook

Now I have the same lean structure as I originally had. The only processor specific code is the file downloads which I think I can live with for now.


Kain0_0‘s answer pointed me in the right direction and helped me simplify it in a way that makes sense.

Two Days with Python & GraphQL


An web application needed to be built. An external API will give me a list of information packets as JSON. The JSON has the information and the user object. The application’s job is to store this data in a local database and provide an user interface to sort and filter this data. Simple enough.

GraphQL kept coming up on on the internet. A number of tools were saying they support GraphQL in their home pages and was making me curious. The requirement also said:

use the technology of your choice REST/GraphQL to build the backend

Now, I had to see what’s it all about. So I sat down read the docs and got a basic understanding of it. It made total sense theoretically. It solved a major problem I face when building Single Page Applications and the Backed REST APIs independently. The opaqueness of incoming data and the right method to get them.

Common Scenario I run into

While building the frontend, we assume use the schema that the backend people give as the source of truth and build it based on that. But the schema becomes stale after a while and changes need to be made. There are many reasons to it:

  • adding/removal/renaming of an attribute
  • optimisations that come into play, which alter the structure
  • the backend API is a third party one and searching and sorting are involved
  • API version changes
  • access control which restricts the information contained..etc.,

And even when we have a stable API, there is the issue of information leak. When you working with user roles, it becomes very confusing very quickly because a request to /user/ returns different objects based on the role of the requester. Admin sees different set of information than a privileged user and a privileged user sees a different set of data than an unprivileged one.

And more often than not, there is a lot of unwanted information that get dumped by APIs on to the frontend than what is required, which sometimes even lead to security issues. If you want to see API response overload take a look under the hood of Twitter web app for example, the API responses have a lot more information than what we see on screen.


Enter GraphQL

GraphQL basically said to me, let’s streamline this process a little bit. First we will stop maintaining resource specific URLs, we are going to just send all our requests to /graphql and that’s it. We won’t be at the mercy of the backend developers whim’s and fancies about how to construct the URL. No more confusing between /course/course_id/lesson/lesson_id/assignments and /assignments?course=course_id&amp;lesson=lesson_id. Next, no, we are not going to use HTTP verbs, everything is just a POST request. And finally no more information overload, you get only what you ask. If you want 3 attributes, then you ask 3, if you want 5 then you ask 5. Let us eliminate the ambiguity and describe what you want as a Graphql document and post it. I mean, I have been sick of seeing SomeObject.someAttribute is undefined errors. So I was willing to put in the effort to define my requests clearly even it meant a little book keeping. Now I will know the exact attributes that I am going to work with. I could filter, sort, paginate all just by defining a query.

It was a breath of fresh air for me. After some hands on experiments I was hooked. This simple app with two types of objects were the perfect candidate to get some experience on the subject.

Day/Iteration 1 – Getting the basic pipeline working

The first iteration went pretty smooth. I found a library called Graphene – Python that implemented GraphQL for Python with support for SQLAlchemy, I added it to Flask with Flask-GraphQL and in almost no time I had a API up and running that will get me the objects, and it came with sorting and pagination. It was wonderful. I was a little confused initially, because, Graphene implements the Relay spec. So my queries looked a little over defined with edges and nodes than plain ones. I just worked with it. I read a quick intro about Connections and realised I didn’t need to worry about it, as I was going to be just querying one object. Whatever implications it had, it was for complex relations.

For the frontend, I added Vue-Apollo the app and I wrote my basic query and the application was displaying data on the web page in no time. It has replaced both Vuex state management and Axios HTTP library in one swoop.

And to help with query designing, there was a helpful auto completing UI called GraphIQL, which was wonderful.

Day/Iteration 2 – Getting search working

Graphene came with sorting and filtering inbuilt. But the filtering is only available if you use Django as it uses django-filter underneath. For SQLAlchemy and Flask, it only offers some tips. Thankfully there was a library called Graphene-SQLAlchemy-Filter which solved this exact problem. I added that and voila, we have a searchable API.

When trying to implement searching in frontend is where things started going sideways. I have to query all the data when loading the page. So the query looked something like

query queryName {
  objectINeeded {
    edges {
      nodes {

And in order to search for something, I needed to do:

query queryName {
  objectINeeded(filters: { attribute_1: "filter_value" }) {

And to sort it would change to:

query queryName {
  objectINeeded(sort: ATTRIBUTE_1_ASC, filters: { attribute_1: "filter_value" }) {

That’s okay for predefined values of sorting and filtering, what if I wanted to do it based on the user input.

1. Sorting

If you notice closely, the sort is not exactly a string I could get from user as an input and frankly it is not even one that I could generate. It is Enum. So I will have to define an ENUM with all the supportable sorts and use that. How do I do that? I will have to define them in a separate GraphQL schema document. I tried doing that and configured webpack to build them and failed miserably. For one, I couldn’t get it to compile the .graphql files. The webloader kept throwing the errors and I lost interest after a while.

2. Searching

The filters is a complex JSON like object that could support OR, AND conditions and everything. I want the values to be based on user input. Apollo supports variables for that purpose. You can do something like this in the Vue script

apollo: {
  myObject: {
    gql: `query GetDataQuery($value1: String, $value2: Int) {
      objectINeed( filters: [{attr1: $value}, {attr2: $value2}] {
    variables() {
      return { value1: this.userInputValue1, value2: this.userInputValue2 }

This is fine when I want to employ both the inputs for searching, what if I want to do only one? Well it turns out I have to define a different query altogether. There is no way to do an optional filter. See the docs on Reactive Queries.
Now that was a lot of Yak shaving I am not willing to do.

Even if I did the Yak Shaving, I ran into trouble on the backend with nested querying. For example what if I wanted to get the objects based on the associated user? Like my query is more like:

query getObjects {
  myObject {
    user(filters: {first_name: "adam"}) {

The Graphene SQLAlchemy documentation said I could do it, it even gave example documentation, but I couldn’t get it working. And when I wanted to implement it myself, the abstraction was too deep that I would have to spend too many hours just doing that.

3. The documentation

The most frustrating part through figuring out all this was the documentation. For some reason GraphQL docs think that if I used Apollo in the frontend, then I must be using Apollo Server in the backend. Turns out there is no strict definition on the semantics for searching/filtering, only on the definition of how to do it. So what the design on the backend should match the design on the frontend. (Now where have I heard that before?) And that’s the reason documentation usually shows both the client and server side implementations.

4. Managing state

An SPA has a state management library like Vuex, Redux to manage application state, but with GraphQL, local state is managed with a GraphQL cache. It improves efficiency by reducing the calls to the server. But here is the catch, you have to define the schema of the objects for that to work. That’s right, define the schema as in write the models in GraphQL documents. It is no big deal if your stack is fully NodeJS, you can just do it once and reference it in both places.

In my case, I will have defined my SQLAlchemy models in Python in the backend, and I will have to do it again in GQL for the frontend. So changes have to be synced between them if anything changes. And remember that each query is defined separately, so I will have to update any query that will be affected by the changes.

At this point I was crying. I has spent close to 8 hours figuring out all this.

I gave up and rewrote the entire freaking app using REST API and finished the project including the UI in the next 6-7 hours and went to bed at 4 in the morning.


  1. GraphQL is a complex solution for a complex problem. You can solve simple problems with it but the complexity will hit you at some point.
  2. It provides a level of clarity in querying data that REST API doesn’t, but it comes with a cost. It is cheap for cheap work and costly for larger requirements. Almost like how AWS bills raise.
  3. No it doesn’t provide the kind of independence between the backend and frontend as it seems like on the surface. This might by lack of understanding and not the goal of GraphQL at all, but if you like me made this assumption, then just know it is invalid.
  4. Use low-level libraries to implement GraphQL, and try to keep it NodeJS. At least for the sake of sharing the schema documents if not for anything. If I has implemented the actions myself instead of depending on Graphene and adding a filter library on top of that, I would have fared better.


The Mastodon ecosystem is really nice. The concept of Fediverse bringing in decentralized social network is a breath of fresh air. I created NiftyBot account in botsin.space – a dedicated server for Mastodon Bots.

What is NiftyBot?

  • It is a Mastodon Bot Account

What does it do?

  • It posts updates about Indian Markets
  • Currently it posts NSE closing report at 4.01 PM everyday. Sample post below


How does it work?

It is a Python script running in AWS Lambda.


A scheduler tiggers the Lambda Function at 4.01 every Monday – Friday. The lambda function is a Python Script that fetches the necessary details from NSE’s output data and posts to Mastodon.

Source Code:

Some asked about if this bot is open source. Obviously, you see the source right here. 🙂 Still I will add the license here.

The above source code is released into the Public Domain. You can do what ever you want with it.

How much does it cost to run this Bot?


Numbers Please:

The AWS Lambda Free tier comes with 1 Million requests and 400,000 GBSec, which is a combination of how much memory we use and the time taken by our process. Since I have used the CloudWatch Scheduler Event as the trigger, I am using 20-22 requests, the Python function takes about 60 MB to run so running at the lowest memory of 128MB block, and usually completes in around 2600-2700 msec. The metrics says my worst billed event so far is about 0.3375 GBSec. With about 20-22 trading days in a month, I might use a total of 8-10 GBSeconds, leaving enough room to run many more bots like this one 🙂

Python Flask என்றால் என்ன?

Python Flask என்பது ஒரு இணைய மென்பொருள்க் குறுங்கட்டமைப்பு.

இணைய மென்பொருள்க் குறுங்கட்டமைப்பா – என்ன சொல்றீங்க?

Flask ஒரு micro web framework. அது ஒரு web application (இணைய செயலி) எழுதத் தேவையான எல்லா வரைமுறைகளையும் நமக்கு வகுத்துத் தருகிறது. அதே சமயம், அப்படி எழுதப்படும் செயலிகள் எல்லாவற்றிலும் எல்லா செயல்பாடுகளும் உள்ளடங்க வேண்டும் என்ற கட்டாயம் இல்லை. நாம் ஒரே ஒரு கோப்பு (file) கொண்டும் ஒரு முழு செயலியை உருவாக்கலாம், அல்லது ஓராயிரம் கோப்புகள் கொண்டும் உருவாக்காலாம். அது முழுக்க முழுக்க நம் தேவையைப் பொருத்தது.

எடுத்துக்காட்டாக, இந்த 👇 நாங்கே வரிகளில் ஒரு முழு செயலியை உருவாக்கிவிட முடியும்.

from flask import Flask

app = Flask(__name__)

def home:
    return "Hello World!"

மேலே உள்ள நிரலை ஒரு கோப்பில் போட்டு நாம் ஒரு webserver-இல் நிறுவி, அதற்கு http://www.sample-flask-application.com என்ற ஒரு முகவரியையும் கொடுத்துவிட்டோம் என்றால், நாம் எப்பொழுதெல்லாம் அந்த இணைய முகவரிக்குப் போகிறோமோ அப்பொழுதெல்லாம் நமக்கு Hello World! என்னும் செய்தி காத்திருக்கும்.

அனால் இணையத்தின் மொழி HTML அல்லவா?

என்னது இது? இணைத்தின் பக்கங்கள் HTML-உம் JavaScript, CSS ஆகிய மொழிகளிலும்தானே எழுதப்படும், இவன் என்ன Pythonஐக் கொண்டு வந்து வைத்துக்கொண்டு பிதற்றுகிறான் என்று நீங்கள் யோசிப்பது எனக்குக் கேட்கிறது. நீங்கள் கூறும் மொழிகள் எல்லாம் இணையப்பக்கங்களின் மொழிகள், அதாவது Client Side. நாம் நம் browserஇல் இணையத்தில் இருக்கும் பக்கங்களைப் பார்க்கப் பயன்படும் மொழிகள். இங்கு நாம் பேசிக்கொண்டு இருக்கும் Python என்பது Server Side மொழி. அதாவது client side வரக்கூடிய பக்கங்களை உருவாக்கித்தரும் செயலியை நாம் எழுத Pythonஐப் பயன்படுத்துகிறோம்.


Python Flaskஐக் கொண்டு, நாம் HTMLஆல் ஆன பக்கங்களை உருவக்க முடியும். அதை எப்படி செய்வது என்பதை அடுத்து அடுத்து வரும் கட்டுரைகளில் பார்க்கலாம். நன்றி.

Creative Commons Licence
This work is licensed under a Creative Commons Attribution-NonCommercial 4.0 International License.

Flask Marshmallow – has no attribute data


Remove .data after the schema.dump() calls in the code. Marshmallow 3 supplies the data directly.

The Issue

If you use Flask-Marshmallow in your Flask application for serialisation of models, then there are chances that you will run into this error when you upgrade dependencies. The reason for this issue is Flask Marshmallow has added support for Marshmallow 3 when moving from version 0.10.0 to 0.10.1. And Marshmallow 3 returns the data directly when the dump function is called instead of creating an object with the .data attribute.

The solution

This is a breaking change and requires the codebase to be updated. Remove all .data accessors from the dump() outputs. For example:

users = user_schema.dump(query_result, many=True).data

# Will become

users = user_schema.dump(query_result, many=True)

Some thoughts

I don’t know why such a big support change is released as a bug fix version 0.10.0 to 0.10.1. It at least should be released as 0.11 in my opinion. If I could go further I would say wrappers for libraries or software in general should always follow the parent’s version number to the minor version. If Marshmallow is in 3.2.x then it makes sense for Flask-Marshmallow to be in 3.2.x. That provides a better idea of what we are using and what are the changes we need to account for.

Parsing & Validating JSON in Flask Requests

This is a follow up to the previous article: Simplifying JSON parsing in Flask routes using decorators

In the previous article we focused on simplifying the JSON parsing using decorators. The aim was to not repeat the same logic in every route adhering to the DRY principle. I will focus on what goes on inside the decorator in this article.

I ended the last article with the following decorator (kindly see the implementation of @required_params in the previous article)

@required_params({"name": str, "age": int, "married": bool})
def ...

where we pass the incoming parameters and their data types and perform their type validation.

Using an external library for validation

The decorator implemented above is suitable for simple use cases. Now, consider the following advanced use cases

  • What if we need more complex validations like Email or Date validation?
  • What if we need to restrict a field to certain values? Say role should be restricted to (teacher, student, admin)?
  • What if need to have custom error messages for each field?
  • What if the value of a parameter is an object with its own set of validation rules?

Solution for neither one of them is going to be trivial enough to be implemented in a single & simple decorator function. This is when the external libraries come to our rescue. Libraries like jsonschema, schematics, Marshmallow ..etc., not only provide functionality, they would also bring more clarity to the codebase, provide modularity and improve the readability.

Note: If you are already using a serialisation library like marshmallow in your project to handle your database Models, you probably knew all this. If you don’t use a serialisation library in your project and instead have something like to_json() or to_dict() function in your models, then you SHOULD consider removing those functions and use a serialisation library.

Example application

Let me layout an example use case which we use to explore this. Here we have a simple app that has two routes which accepts JSON payload. The expected JSON data is described in the decorator @required_params(...) and the validation is carried out inside the decorator function.

Requests and responses

Now let us sent an actual request to this application.


Now that is really nice, we have got a “201 Created” response. Let us now try with a wrong input. I am setting the married field to twice instead of true or false, the expected boolean values.


That returns a 400 Bad request as expected. The decorator has tried validating the types of the input and has found that one of the values is not in the expected format. But the error message itself is kind of crude:

  1. It doesn’t actually tell us which parameter is wrong. In a complex object this might lead to wasting a lot of time jumping between the docs and the input to guess and find the wrong parameter.
  2. The data types are represented as Python class types like class 'int'. While this might convey the intended meaning, it is still far better to say something decent like integer instead.

Using Marshmallow for validation

Using the marshmallow we can define the schema of our expected JSON.

from marshmallow import Schema, fields, ValidationError

class UserSchema(Schema):
    first_name = fields.String(required=True)
    last_name = fields.String(required=True)
    age = fields.Integer(required=True)
    married = fields.Boolean(required=True)

Now that the marshmallow will take care of the validation, we can update our decorator too:

def required_params(schema):
    def decorator(fn):

        def wrapper(*args, **kwargs):
            except ValidationError as err:
                error = {
                    "status": "error",
                    "messages": err.messages
                return jsonify(error), 400
            return fn(*args, **kwargs)

        return wrapper
    return decorator

And finally we pass an object of the UserSchema instead of a list of params in the decorator:

@app.route("/user/", methods=["POST"])
def add_user():
    # here a simple list is used in place of a DB
    return "OK", 201

Note: I have passed the strict=True so that Marshmallow will raise a ValidationError. By default marshmallow 2 doesn’t raise an error, however, it should be happening in version 3. Check your version to see if the strict parameter is necessary.

With the app updated, now let us send a request and test it.


Good. We get the “not a boolean” validation error. Now what if we have multiple errors?


Sweet, we get parameter specific error messages even for multiple errors. If you remember our original implementation, only one error could be returned at a time because the first exception would return a response. Using the library provides a good upgrade from that.

By defining multiple schemas for the various data models that we expect as an input, we could perform complex validations independent of our view function. This gives us clean view functions that handle just the business logic.


This post only exposes the basic implementation of using the libraries and simplifying parsing and validation of incoming JSON. The marshmallow library offers a lot more like: complex validators like Email and Date, parsing subset of a schema by using only, custom validators (age between 30-35 for example), loading the data directly into SQLAlchemy models (check out Flask-marshmallow)..etc., These can really make app development easier, safer and faster.

Schematic is another library offering similar functionalities that focuses on ORM integration.