Async

We can run asyncio.gather() in the background by not awaiting the call to asyncio.gather(). The asyncio.gather() returns an asyncio.Future that does not have to be awaited. It is scheduled for execution in the asyncio event loop, along with all coroutines provided to the gather. Therefore, the caller is free to proceed with other activities and […]

We can run an asyncio event loop in a new thread by starting a new thread and configuring it to start or run an event loop. There are many approaches we can use to run an event loop in a new thread. The simplest is to configure a new thread to start an event loop […]

We can automatically handle exceptions in coroutines executed via asyncio.gather() by setting the “return_exceptions” argument to True. By default, if a coroutine is executed by asyncio.gather() fails with an unhandled exception, it will be propagated to the caller. Setting the “return_exceptions” argument to True will trap any unhandled exceptions and provide them as return values, […]

Understanding Concurrency in Python Concurrency refers to the execution of multiple tasks simultaneously in a program. There are primarily three ways to introduce concurrency in Python - Multithreading, Multiprocessing and Asyncio.

Intro to using Python's httpx library for web scraping. Proxy and user agent rotation and common web scraping challenges, tips and tricks.

One of the main appeals of using Python’s asyncio is being able to fire off many coroutines and run them concurrently. How many ways do you know for waiting for their results?

In this video, Ben Finkel covers how to create an asynchronous API call in Python. A lot of internet resources have detailed articles that discuss all the in...

Learn how to speed up slow Python code with concurrent programming and the cutting-edge asyncio library./bbr/br/ Python is flexible, versatile, and easy to learn. It can also be very slow compared to lower-level languages. Python Concurrency with asyncio/i teaches you how to boost Python's performance by applying a variety of concurrency techniques. You'll learn how the complex-but-powerful asyncio library can achieve concurrency with just a single thread and use asyncio's APIs to run multiple web requests and database queries simultaneously. The book covers using asyncio with the entire Python concurrency landscape, including multiprocessing and multithreading.

05bit/peewee-async, peewee-async Asynchronous interface for peewee ORM powered by asyncio. Important notes Since version 0.6.0a only peewee 3.5+ is supported If you still

In this video, I will show you how to take a slow running script with many API calls and convert it to an async version that will run much faster. I use AIOH...

Threading in Python is simple. It allows you to manage concurrent threads doing work at the same time. The library is called "threading", you create "Thread" objects, and they run target functions for you. You can start potentially hundreds of threads that will operate in parallel, and work through tasks faster.

Easily learn how to download multiple images in Python using synchronous and asynchronous code.

We can run multiple concurrent asyncio event loops by starting and running each new event loop in a separate thread. Each thread can host and manage one event loop. This means we can start one thread per event loop we require, allowing a program to potentially scale from thousands to millions of coroutines. In this […]

We can define coroutine methods on custom Python objects. This allows methods on custom Python objects to use async/await syntax, such as awaiting other coroutines and tasks and allows the custom coroutine methods themselves to be awaited within our asyncio programs. In this tutorial, you will discover how to define object methods as coroutines. Let’s […]

This tutorial will give you a firm grasp of Python’s approach to async IO, which is a concurrent programming design that has received dedicated support in Python, evolving rapidly from Python 3.4 through 3.7 (and probably beyond).

This is a collection of documents published by the BBC R&D Cloudfit team delving into the detail of technologies we have found interesting or useful during the course of our work.

Asyncio allows us to use asynchronous programming with coroutine-based concurrency in Python. Although asyncio has been available in Python for many years now, it remains one of the most interestin…

The Thread class allows you to create and manage new threads of execution in Python. Although the threads have been available since Python 2, it is not widely used, perhaps because of misunderstand…

I do a lot of web scraping in my spare time, and have been chasing down different formats and code snippets to make a large amount of network requests locally, with controls for rate limiting and error handling.

A guide to writing more responsive and performant programs

In this tutorial we will implement a Python based FastAPI with PostgreSQL CRUD. We will focus on implementing Asynchronous REST Endpoints with databases asyncio

Twilio posts cloud communications trends, customer stories, and tips for building scalable voice and SMS applications with Twilio's APIs.

This tutorial looks at how to configure SQLAlchemy, SQLModel, and Alembic to work with FastAPI asynchronously.

How to retrieve LIVE cryptocurrency prices

Peewee Tutorial, Peewee is a Python ORM (Object-Relational Mapping) library which supports SQLite, MySQL, PostgreSQL and Cockroach databases. This tutorial will help you to unde

This post is part of the FastAPI series. In the first post, I introduced you to FastAPI and how you can create high-performance Python-based applications in it. In this post, we are going to work on Rest APIs that interact with a MySQL DB. We will also be looking at how we can organize routers and models in multiple files to make them maintainable and easier to read. FastAPI does not strict you to use a certain database framework. You may use SQLAlchemy or any other you want. I’d prefer to use peewee since it’s more expressible and easier to use. Installing Peewee and MySQL Drivers Before start using MySQL in our app we must have all things at our end ready. We will first install peewee ORM for that purpose. I will create a separate file called database.py and I will be adding the following code to it. from peewee import * user = 'root' password = 'root' db_name = 'fastapi_contact' conn = MySQLDatabase( db_name, user=user, password=password, host='localhost' ) class BaseModel(Model): class Meta: database = conn I imported the framework and created a MySQLDatabase object. I am passing all required credentials to make it a successful connection. But wait… before we run the server, we also need to install the MySQL driver. I will be using PyMySQL to connect the MySQL database otherwise you will be getting the error: self._state.set_connection(self._connect()) File "/Users/AdnanAhmad/.local/share/virtualenvs/ContactAPI-Ycq5CXn7/lib/python3.9/site-packages/peewee.py", line 3964, in _connect raise ImproperlyConfigured('MySQL driver not installed!') peewee.ImproperlyConfigured: MySQL driver not installed! So far so good. Now I am going to change the main.py to import database.py file contents. app = FastAPI(title='Contact.ly', description='APIs for contact Apis', version='0.1') @app.on_event("startup") async def startup(): print("Connecting...") @app.get("/") async def root(): return {"message": "Contact Applications!"} A few things happening but let me discuss the important one first: FastAPI events. FastAPI Events FastAPI comes up with a couple of events that you can use in your apps: Startup and Shutdown. Startup Event: This event is responsible to carry out certain tasks while starting the application. By starting the application means that when you hit a certain API endpoint, it gets executed first. Tasks like connecting to the database are one of the tasks necessary to be performed before doing anything else. Shutdown Event: This event is triggered when the app is quitting or shutting down. You might like to use this event to store all info before leaving the app. The advantage of these events is that you do not have to call the code in every API endpoint. So as you understood by now, I am connecting the DB on the startup event. The other thing you would have noticed that I passed some named parameters in FastAPI constructor. when I do that, it produces something like below: Looks much better, No? Now I have to create routers, models, DB connections, etc. These all in a single file will make the app bloated. I am going to create a couple of folders right now with the name models and routers respectively. I am also creating a file database.py in the root folder that will have instantiate MySQLDatabase class object. What is APIRouter It is OK to put all your routes in a single main.py file if you have a small app but what if you are working on a big system then putting all routes in a single file would make it clumsy and bloated. How nice would it be if we can organize a set of routers as per entity or apps? Something like Django offers. FastAPI’s APIRouter can help you to organize your routes and logic in multiple files. In the root folder, I created a folder named routers where I will be adding my contact API related routers. I am adding an __init.py__ in it to turn it into a package. You are free not to do it. I am adding my route file, contact.py in it with the following content: from fastapi import APIRouter router_contacts = APIRouter( prefix="/contacts", tags=["contacts"] ) @router_contacts.get("/",summary="List of contacts", description="Returns all contacts" ) async def get_contacts(): #create(first_name='Addu', last_name='Pagal', email='addu@gmail.com', phone='123-494', status=1) return [{'status': 'OK'}] I created an APIRouter object by passing two parameters: prefix and tags. This frees us to add /contacts/ in every end-point. I also added summary and description parameters which are rendered in docs. FastAPI also lets you use docstring to document your API endpoints. For instance: @router_contacts.get("/view/{id}", summary="Returns a single contact") async def view(id: int): """ To view all details related to a single contact - **id**: The integer id of the contact you want to view details. """ return [{'status': 'OK'}] Here I am using docstring with markdown. When it’s rendered it is displayed as: Cool, No? Docstring gives you more flexibility on how do you want to display your API endpoint documentation. Tags help to organize app-related APIs at a single place in docs. They are also used in OpenAPI. For instance when I visit http://localhost:8080/openapi.json it shows the following where you can see tags are given there. {"openapi":"3.0.2","info":{"title":"Contact.ly","description":"APIs for contact Apis","version":"0.1"},"paths":{"/":{"get":{"summary":"Root","operationId":"root__get","responses":{"200":{"description":"Successful Response","content":{"application/json":{"schema":{}}}}}}},"/contacts/":{"get":{"tags":["contacts"],"summary":"List of contacts","description":"Returns all contacts","operationId":"get_contacts_contacts__get","responses":{"200":{"description":"Successful Response","content":{"application/json":{"schema":{}}}}}}},"/contacts/view/{id}":{"get":{"tags":["contacts"],"summary":"View","description":"Returns a single contact","operationId":"view_contacts_view__id__get","parameters":[{"required":true,"schema":{"title":"Id","type":"integer"},"name":"id","in":"path"}],"responses":{"200":{"description":"Successful Response","content":{"application/json":{"schema":{}}}},"422":{"description":"Validation Error","content":{"application/json":{"schema":{"$ref":"#/components/schemas/HTTPValidationError"}}}}}}}},"components":{"schemas":{"HTTPValidationError":{"title":"HTTPValidationError","type":"object","properties":{"detail":{"title":"Detail","type":"array","items":{"$ref":"#/components/schemas/ValidationError"}}}},"ValidationError":{"title":"ValidationError","required":["loc","msg","type"],"type":"object","properties":{"loc":{"title":"Location","type":"array","items":{"type":"string"}},"msg":{"title":"Message","type":"string"},"type":{"title":"Error Type","type":"string"}}}}}} We have created our routes file but things are not done yet. We have to call them in the main.py file. In the main file, I am going to make the following changes. from routers import contact app = FastAPI(title='Contact.ly', description='APIs for contact Apis', version='0.1') app.include_router(contact.router_contacts) Since routers folder is a package now, I can call the router_contacts variable and pass it to include_router function. After doing the new contacts related routes will be appearing on the documentation page. Peewee Models As I mentioned I am going to use peewee ORM instead of SqlAlchemy, I am going to create relevant models for this purpose. Before I do that, let’s take a look at what database.py file will look like. This file will contain the class object for MySQLDatabase class having all the required parameters passed to it. The database.py will look like below: from peewee import * user = 'root' password = 'root' db_name = 'fastapi_contact' conn = MySQLDatabase( db_name, user=user, password=password, host='localhost' ) Later, in the main.py file, I am going to call this like: from database import * @app.on_event("startup") async def startup(): if conn.is_closed(): conn.connect() @app.on_event("shutdown") async def shutdown(): print("Closing...") if not conn.is_closed(): conn.close() So whenever any API-endpoint will be accessed, it’d create a connection and then close it once all execution is done. Now models… I will be creating a couple of models: A BaseModel that will be responsible for associating models with MySQL DB and a Contact model dealing with the contacts table. I am assuming that table is already created hence not covering it. To learn more about it you may check my article here. from peewee import * from database import conn class BaseModel(Model): class Meta: database = conn and contact.py from peewee import * from .Base import BaseModel class Contact(BaseModel): first_name = CharField(max_length=30) last_name = CharField(max_length=30) email = CharField(max_length=40) phone = CharField(max_length=25) status = SmallIntegerField() updated_at = DateTimeField() class Meta: db_table = 'contacts' We are going to create CRUD endpoints to manage the entire app that is API endpoints for creating a record, updating, deleting, and retrieving. I am going to create our very first contact API endpoint which will be adding a contact in the DB. In models/contact.py I am going to create a method for data insertion async def create_contact(first_name: str, last_name: str, email: str, phone: str, status: int): contact_object = Contact( first_name=first_name, last_name=last_name, email=email, phone=phone, status=status ) contact_object.save() return contact_object Pretty straight forward. I created a Contact object and passed the required parameters in it. Now pay attention. The create_contact model method is returning a ModelSelect object which is not useful for us as it contains several peewee-related information hence, we need a Pydantic model to take care of it. It is not so straightforward so let’s dive in! First, I am going to create a class that will map peewee models to Pydantic models. The code has been taken from the FastAPI official website: class PeeweeGetterDict(GetterDict): def get(se...