Pymetrix
Introduction:
Pymetrix (previously known as Analyx) is a plug-and-play analytics library written in Python. It aims to be lightweight and effective at capturing most of the basic metrics for your website, with the possibility of adding more through the use of extensions.
How to use it
Pymetrix is really easy to integrate with your projects. Here's an example:
Let's say you want to monitor a method foo()
defined as:
After adding the required lines, the code will look something like this:
from random import randint
from pymetrix.metrics import Metrics
metricman = Metrics(loc=__file__)
foo_obj = None
def foo():
print(f"Hello world {randint(0,1000000)}!")
if foo_obj is None:
ep1 = endpoints.Endpoint(id="/", endpoint=foo)
foo_obj = flow.FlowNode(ep1, name="Object1")
metricman.add_to_analytics(foo_obj, layerName="foo")
Using with various Web Frameworks
Pymetrix can be used in conjunction with other frameworks as well, including web frameworks. The following examples will illustrate it more clearly.
1. Using with Django
Django is the most popular framework of choice for a vast majority of Python web developers, often dubbed as "the framework for perfectionists with deadlines". It is a clean, concise and opinionated framework that enforces code reusability, modularity and code readability. It has an intuitive directory structure, and a large ecosystem of plugins tailor-made for it. Besides those, the usual Python libraries that can be used in tandem with it for designing business logic as well.
Being a library, you can import Pymetrix on any file that will be executed by the server on calling the endpoints. This is usually the views.py
file in a standard Django project.
A sample views.py
in Django with Pymetrix (Tested on Django 4.0.0):
from django.shortcuts import render, HttpResponse
from django.http import StreamingHttpResponse
from typing import Dict, List, Any, Tuple, NewType
from pymetrix import metrics, endpoints, flow
import pymetrix.visualize as vs
# Create your views here.
metricman = metrics.Metrics(loc="Test") # Initialize the Pymetrix Metrics object
def blog(request):
# Other code
...
node2: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/blog", endpoint=blog), name="Blog"
)
metricman.add_to_analytics(
node2
) # Add the function to the graph corresponding to the metricman object
# any return stuff
return HttpResponse("<h1>Blog works!</h1>")
def about(request):
# Other code
...
node3: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/about", endpoint=about), name="About"
)
metricman.add_to_analytics(
node3
) # Add the function to the graph corresponding to the metricman object
# any return stuff
return HttpResponse("<h1>About works!</h1>")
def contact(request):
# Other code
...
node4: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/contact", endpoint=contact), name="Contact"
)
metricman.add_to_analytics(
node4
) # Add the function to the graph corresponding to the metricman object
# any return stuff
return HttpResponse("<h1>Contact works!</h1>")
def collaborations(request):
# Other code
...
node6: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/collaborations", endpoint=collaborations),
name="Collaborations",
)
metricman.add_to_analytics(
node6
) # Add the function to the graph corresponding to the metricman object
# any return stuff
return HttpResponse("<h1>Collaborations works!</h1>")
def events(request):
# Other code
...
node5: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/events", endpoint=events), name="Events"
)
metricman.add_to_analytics(
node5
) # Add the function to the graph corresponding to the metricman object
# any return stuff
return HttpResponse("<h1>Events works!</h1>")
def home(request):
node1: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/home1", endpoint=home), name="Home"
)
metricman.add_to_analytics(
node1
) # Add the function to the graph corresponding to the metricman object
return HttpResponse("<h1>Home works!</h1>")
# The methods below won't be added for metrics
def someMethod(request):
...
class someClass:
...
def anotherMethod(request):
...
def aThirdMethod(request):
...
Although I haven't tested it yet, owing to the nature of the metricman
object, I think it can be used in the top level urls.py
for being used over the entire project - one that has multiple Django Apps into it.
Also, do check out Pymetrix Django Dashboard
2. Using with FastAPI
FastAPI is a micro web framework built on top of Pydantic and Starlette. It claims to be the fastest Pythonic web framework out there for building APIs, and adheres to the OpenAPI standards. It provides two UIs for testing the APIs made through it - Redoc and SwaggerUI. It also provides a lot of functionalities out of the box, such as asynchronous calls, which speeds up API calls, sometimes tremendously. It can be used both with ASGI and WSGI.
Integrating Pymetrix with FastAPI is as easy as it is in Django, if not easier.
from fastapi import FastAPI, status
from pymetrix import metrics, endpoints, flow
from fastapi.responses import ORJSONResponse
app = FastAPI()
metricman = metrics.Metrics(loc="Test")
@app.get("/home", status_code=status.HTTP_200_OK, response_class=ORJSONResponse)
async def home():
node1: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/home", endpoint=home), name="Home"
)
metricman.add_to_analytics(node1)
return {"response": status.HTTP_200_OK, "message": "This is the home page"}
@app.get("/contact", status_code=status.HTTP_200_OK, response_class=ORJSONResponse)
async def contact():
node2: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/contact", endpoint=contact), name="Contact"
)
metricman.add_to_analytics(node2)
return {"response": status.HTTP_200_OK, "message": "This is the contact page"}
@app.get("/collaborations", status_code=status.HTTP_200_OK, response_class=ORJSONResponse)
async def collaborations():
node3: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/collaborations", endpoint=collaborations),
name="Collaborations",
)
metricman.add_to_analytics(node3)
return {
"response": status.HTTP_200_OK,
"message": "This is the collaborations page",
}
@app.get("/events", status_code=status.HTTP_200_OK, response_class=ORJSONResponse)
async def events():
node4: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/events", endpoint=events), name="Events"
)
metricman.add_to_analytics(node4)
return {"response": status.HTTP_200_OK, "message": "This is the events page"}
@app.get("/blog", status_code=status.HTTP_200_OK, response_class=ORJSONResponse)
async def blog():
node5: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/blog", endpoint=blog), name="Blog"
)
metricman.add_to_analytics(node5)
return {"response": status.HTTP_200_OK, "message": "This is the blog page"}
@app.get("/about", status_code=status.HTTP_200_OK, response_class=ORJSONResponse)
async def about():
node6: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(id="/about", endpoint=about), name="About"
)
metricman.add_to_analytics(node6)
return {"response": status.HTTP_200_OK, "message": "This is the about page"}
@app.get("/statistics", status_code=status.HTTP_200_OK, response_class=ORJSONResponse)
async def get_statistics():
return {"response": status.HTTP_200_OK, "message": metricman.aggregate()}
Now we start the uvicorn
server using:
Now going to localhost:8000/statistics
will give you the aggregate hits on each endpoint.
Also, here we are serializing the data using orjson, instead of regular JSON
Just like the case with Django, you can initialize a Pymetrix object in the main.py file and then import into other scripts from there, if you want a centralized view. But you're also free to take a modular approach with your project by initializing the Pymetrix objects within each module - your skills are your limit.
3. Using with Starlite
Starlite is the new challenger in town for Web API frameworks. It's similar to FastAPI in the aspect that it's based on Starlette and Pydantic. But it has a fundamental philosophical difference - Starlite is opinionated, while FastAPI is not.
According to the maker of Starlite, Na'aman Hirschfeld (who is also an online friend of mine 😎):
The intention behind Starlite was to create a higher level opinionated API framework. I placed opinionated in bold because in my view, being opinionated regarding how certain things should be done and shouldn’t be done, and establishing best practices, is one of the most important things a framework can do.
So let's see how Pymetrix works with Starlite:
from starlite import Starlite, get
from pymetrix import metrics, endpoints, flow
from typing import Dict
metricman = metrics.Metrics(loc="Test")
@get(path="/home")
def home() -> Dict:
node1: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(endpoint="/home", id=home), name="Home"
)
metricman.add_to_analytics(node1)
return {"response": 200, "message": "Home Page Works"}
@get(path="/contact")
def contact() -> Dict:
node2: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(endpoint="/contact", id=contact), name="Contact"
)
metricman.add_to_analytics(node2)
return {"response": 200, "message": "Contact Page Works"}
@get(path="/collaborations")
def collaborations() -> Dict:
node3: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(endpoint="/collaborations", id=collaborations),
name="Collaborations",
)
metricman.add_to_analytics(node3)
return {"response": 200, "message": "Collaborations Page Works"}
@get(path="/events")
def events() -> Dict:
node4: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(endpoint="/events", id=events), name="Events"
)
metricman.add_to_analytics(node4)
return {"response": 200, "message": "Events Page Works"}
@get(path="/blog")
def blog() -> Dict:
node5: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(endpoint="/blog", id=blog), name="Blog"
)
metricman.add_to_analytics(node5)
return {"response": 200, "message": "Blog Page Works"}
@get(path="/about")
def about() -> Dict:
node6: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(endpoint="/about", id=about), name="About"
)
metricman.add_to_analytics(node6)
return {"response": 200, "message": "About Page Works"}
@get(path="/statistics")
def statistics() -> Dict:
node2: flow.FlowNodeType = flow.FlowNode(
endpoints.Endpoint(endpoint="/contact", id=contact), name="Contact"
)
metricman.add_to_analytics(node2)
return {"response": 200, "message": metricman.time_series()}
app = Starlite(
route_handlers=[home, contact, collaborations, events, blog, about, statistics]
)
Do note that Starlite uses the orjson library for its responses by default.
Orjson Note
In orjson, the response is a JSON byte string and not a object. This dramatically speeds up the serialization process.