Python Falcon – Sqlalchemy Models

Python Falcon – SQLAlchemy Models ”; Previous Next To demonstrate how the Falcon”s responder functions (on_post(), on_get(), on_put() and on_delete()), we had done CRUD (which stands for Create, Retrieve, Update and Delete) operations on an in-memory database in the form of a Python list of dictionary objects. Instead, we can use any relational database (such as MySQL, Oracle etc.) to perform store, retrieve, update and delete operations. Instead of using a DB-API compliant database driver, we shall use SQLAlchemy as an interface between Python code and a database (we are going to use SQLite database as Python has in-built support for it). SQLAlchemy is a popular SQL toolkit and Object Relational Mapper. Object Relational Mapping is a programming technique for converting data between incompatible type systems in object-oriented programming languages. Usually, the type system used in an Object Oriented language like Python contains non-scalar types. However, data types in most of the database products such as Oracle, MySQL, etc., are of primitive types such as integers and strings. In an ORM system, each class maps to a table in the underlying database. Instead of writing tedious database interfacing code yourself, an ORM takes care of these issues for you while you can focus on programming the logics of the system. In order to use SQLALchemy, we need to first install the library using PIP installer. pip install sqlalchemy SQLAlchemy is designed to operate with a DBAPI implementation built for a particular database. It uses dialect system to communicate with various types of DBAPI implementations and databases. All dialects require that an appropriate DBAPI driver is installed. The following are the dialects included − Firebird Microsoft SQL Server MySQL Oracle PostgreSQL SQLite Sybase Database Engine Since we are going to use SQLite database, we need to create a database engine for our database called test.db. Import create_engine() function from sqlalchemy module. from sqlalchemy import create_engine from sqlalchemy.dialects.sqlite import * SQLALCHEMY_DATABASE_URL = “sqlite:///./test.db” engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args = {“check_same_thread”: False}) In order to interact with the database, we need to obtain its handle. A session object is the handle to database. Session class is defined using sessionmaker() – a configurable session factory method which is bound to the engine object. from sqlalchemy.orm import sessionmaker, Session session = sessionmaker(autocommit=False, autoflush=False, bind=engine) Next, we need a declarative base class that stores a catalog of classes and mapped tables in the Declarative system. from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() Model class Students, a subclass of Base is mapped to a students table in the database. Attributes in the Books class correspond to the data types of the columns in the target table. Note that the id attribute corresponds to the primary key in the book table. class Students(Base): __tablename__ = ”student” id = Column(Integer, primary_key=True, nullable=False) name = Column(String(63), unique=True) marks = Column(Integer) Base.metadata.create_all(bind=engine) The create_all() method creates the corresponding tables in the database. It can be confirmed by using a SQLite Visual tool such as SQLiteStudio. We now need to declare a StudentResource class in which the HTTP responder methods are defined to perform CRUD operations on students table. The object of this class is associated to routes as shown in the following snippet − import falcon import json from waitress import serve class StudentResource: def on_get(self, req, resp): pass def on_post(self, req, resp): pass def on_put_student(self, req, resp, id): pass def on_delete_student(self, req, resp, id): pass app = falcon.App() app.add_route(“/students”, StudentResource()) app.add_route(“/students/{id:int}”, StudentResource(), suffix=”student”) on_post() Rest of the code is just similar to in-memory CRUD operations, with the difference being the operation functions interact with the database through SQLalchemy interface. The on_post() responder method first constructs an object of Students class from the request parameters and adds it the Students model. Since this model is mapped to the students table in the database, corresponding row is added. The on_post() method is as follows − def on_post(self, req, resp): data = json.load(req.bounded_stream) student=Students(id=data[”id”], name=data[”name”], marks=data[”marks”]) session.add(student) session.commit() resp.text = “Student added successfully.” resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_TEXT As mentioned earlier, the on_post() responder is invoked when a POST request is received. We shall use Postman app to pass the POST request. Start Postman, select POST method and pass the values (id=1, name=”Manan” and marks=760 as the body parameters. The request is processed successfully and a row is added to the students table. Go ahead and send multiple POST requests to add records. on_get() This responder is meant to retrieve all the objects in the Students model. query() method on Session object retrieves the objects. rows = session.query(Students).all() Since the default response of Falcon responder is in JSON format, we have to convert the result of above query in a list of dict objects. data=[] for row in rows: data.append({“id”:row.id, “name”:row.name, “marks”:row.marks}) In the StudentResource class, let us add the on_get() method that performs this operation and sends its JSON response as follows − def on_get(self, req, resp): rows = session.query(Students).all() data=[] for row in rows: data.append({“id”:row.id, “name”:row.name, “marks”:row.marks}) resp.text = json.dumps(data) resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_JSON The GET request operation can be tested in the Postman app. The /students URL will result in displaying JSON response showing data of all objects in the students model. The two records shown in the result pane of Postman app can also be verified in the data view of SQLiteStudio. on_put() The on_put() responder performs the UPDATE operation. It responds to the URL /students/id. To fetch the object with given id from the Students model, we apply the filter to the query result, and update the values of its attributes with the data received from the client.

Python Falcon – Testing

Python Falcon – Testing ”; Previous Next Falcon”s testing module is a Functional testing framework for Falcon apps. It contains various test classes and utility functions to support functional testing. The testing framework supports both unittest and pytest. We shall use following the script (myapp.py) to demonstrate testing functionality. It contains a HelloResource class with an on_get() responder that renders a JSON response of Hello World. The create() function returns Falcon”s Application object added with a route registered with ”/” URL. from waitress import serve import falcon import json class HelloResource: def on_get(self, req, resp): “””Handles GET requests””” resp.text=json.dumps({“message”:”Hello World”}) # This is the default status resp.status = falcon.HTTP_200 # Default is JSON, so override resp.content_type = falcon.MEDIA_JSON def create(): app = falcon.App() hello = HelloResource() app.add_route(”/”, hello) return app app=create() if __name__ == ”__main__”: serve(app, host=”0.0.0.0”, port=8000) Using unittest The testing.TestCase extends unittest to facilitate functional testing of WSGI/ASGI applications written with Falcon. We need to inherit from this base class and write the tests. The test functions in the TestCase subclass are of the name simulate_*() where ”*” stands for HTTP methods like GET, POST etc. It means, we have to fetch the result of simulate_get() function and compare it with the expected result by assertion functions. The simulate_*() functions receive two arguments. simulate_*(app, route) Following is the code for test-myapp.py. It executes simulate_get() function and asserts its result with the anticipated result and indicates whether the test has failed or passed. from falcon import testing import myapp class MyTestCase(testing.TestCase): def setUp(self): super(MyTestCase, self).setUp() self.app = myapp.create() class TestMyApp(MyTestCase): def test_get_message(self): doc = {”message”: ”Hello world!”} result = self.simulate_get(”/”) self.assertEqual(result.json, doc) if ”__name__”==”__main__”: unittest.main() Run the above test with the help of the following command − python -m unittest test-myapp.py F ============================================================== FAIL: test_get_message (test-myapp.TestMyApp) ————————————————————– Traceback (most recent call last): File “E:falconenvtest-myapp.py”, line 17, in test_get_message self.assertEqual(result.json, doc) AssertionError: {”message”: ”Hello World”} != {”message”: ”Hello world!”} – {”message”: ”Hello World”} ? ^ + {”message”: ”Hello world!”} ? ^ + ————————————————————– Ran 1 test in 0.019s FAILED (failures=1) Using Pytest To perform testing using PyTest framework, you need to install it using PIP utility. pip3 install pytest To run a test function, we need an object of testing.TestClient class. It simulates the requests for WSGI and ASGI applications. This object is first obtained by giving Falcon application object as the argument. We run the simulate_*() functions and assert its result with the anticipated output to decide whether the test has failed or passed. In both the examples, the test fails because of difference in case of ”W” in Hello World message. The responder returns it with uppercase ”W” whereas the test function has it in lowercase. from falcon import testing import pytest import myapp @pytest.fixture() def client(): return testing.TestClient(myapp.create()) def test_get_message(client): doc = {”message”: ”Hello world!”} result = client.simulate_get(”/”) assert result.json == doc Run the above test using the following command − pytest test-myapp.py –v =========== test session starts ========================== platform win32 — Python 3.8.6, pytest-7.1.2, pluggy-1.0.0 – e:falconenvscriptspython.exe cachedir: .pytest_cache rootdir: E:falconenv plugins: anyio-3.5.0 collected 1 item test-myapp.py::test_get_message FAILED [100%] ==================== FAILURES ======================= _____________________________________________________ test_get_message _____________________________________________________ client = <falcon.testing.client.TestClient object at 0x0000000003EAA6A0> def test_get_message(client): doc = {”message”: ”Hello world!”} result = client.simulate_get(”/”) > assert result.json == doc E AssertionError: assert {”message”: ”Hello World”} == {”message”: ”Hello world!”} E Differing items: E {”message”: ”Hello World”} != {”message”: ”Hello world!”} E Full diff: E – {”message”: ”Hello world!”} E ? ^ - E + {”message”: ”Hello World”} E ? ^ test-myapp.py:42: AssertionError ============ short test summary info ================== FAILED test-myapp.py::test_get_message – AssertionError: assert {”message”: ”Hello World”} == {”message”: ”Hello world!”} ============ 1 failed in 4.11s ======================== Print Page Previous Next Advertisements ”;

Python Falcon – Error Handling

Python Falcon – Error Handling ”; Previous Next To handle various error situations, the above status codes can be used for the response object. Falcon also provides set of error classes. Their object can be raised when corresponding runtime error situation arises. These error classes are derived from HTTPError class as their base class. The error object is raised as shown in the following example − import falcon class MyResource: def on_get(self, req, resp): # some Python code raise falcon.HTTPBadRequest( title=”Value Out of Range”, description=”The value is not between permissible range” ) Predefined Error Classes Some of the predefined error classes provided by Falcon are as follows − HTTPBadRequest − 400 Bad Request. The server is unable to process the request due to a client error such as malformed request syntax, invalid request message framing etc. HTTPInvalidHeader − Results in 400 Bad Request because one of the headers in the request is invalid. HTTPInvalidParam − Represents 400 Bad Request. This error may refer to an invalid parameter in a query string, form, or document that was submitted with the request. HTTPMissingParam − 00 Bad Request is raised when a parameter is missing from the request. HTTPForbidden − The server understood the request but refuses to authorize it. The status code is 403 Forbidden. HTTPNotFound − When the server did not find a current representation for the target resource, a 404 status code is raised. It does not indicate whether this lack of representation is temporary or permanent. HTTPMethodNotAllowed − 405 Method Not Allowed. The method received in the request-line is not supported by the target resource. HTTPLengthRequired − When The server refuses to accept the request without a defined Content- Length. 411 Length Required. Error code. HTTPUnsupportedMediaType − If the origin server is refusing to service the request because the payload is in a format not supported by this method on the target resource. Equivalent status code is 415 Unsupported Media Type. HTTPUnprocessableEntity − If the server understands the content type of the request entity and the syntax of the request entity is correct but was unable to process the contained instructions, the error status code raised is 422 Unprocessable Entity. For example, if an XML request body contains well-formed, but semantically erroneous, XML instructions. HTTPTooManyRequests − A 429 Too Many Requests status code is raised when the user has sent too many requests in a given amount of time (“rate limiting”). HTTPInternalServerError − A very common error situation resulting in 500 Internal Server Error. The server encountered an unexpected condition that prevented it from fulfilling the request. HTTPNotImplemented − The 501 (Not Implemented) status code indicates that the server does not support the functionality required to fulfill the request. This is the appropriate response when the server does not recognize the request method and is not capable of supporting it for any resource. HTTPServiceUnavailable − 503 Service Unavailable means that the server is currently unable to handle the request due to a temporary overload or scheduled maintenance. MediaNotFoundError − 400 Bad Request. This Exception is raised by a media handler when trying to parse an empty body. MediaMalformedError − 400 Bad Request. This Exception is raised by a media handler when trying to parse a malformed body. Redirection There are also a set of exceptions, which when raised, trigger a redirection response to the client. The status codes are of the type 3xx. These exceptions, represented by following classes, shot-circuit the request processing as a subclass of HttpError. HTTPMovedPermanently − 301 Moved Permanently. This status code indicates that the target resource has been assigned a new permanent URI. HTTPFound − 302 Found status code meaning that the target resource resides temporarily under a different URI. HTTPTemporaryRedirect − This class raises the 307 (Temporary Redirect) status code, which means that the target resource resides temporarily under a different URI and the user agent MUST NOT change the request method if it performs an automatic redirection to that URI. HTTPPermanentRedirect − Results ib 308 Permanent Redirect, indicating that the target resource has been assigned a new permanent URI. Print Page Previous Next Advertisements ”;

Python Falcon – Deployment

Python Falcon – Deployment ”; Previous Next It is possible to use Apache server enabled with the mod_wsgi module to deploy a Falcon web app, just as any WSGI app. Another alternative is to use uWSGI or gunicorn for deployment. The uWSGI is a fast and highly configurable WSGI server. If used along with NGNIX, it gives better performance in the form of speed in the production ready environment. First, install Falcon and uWSGI in a Python virtual environment with PIP installer and expose the Falcon”s application object to uWSGI it with wsgi.py as below − import os import myapp config = myproject.get_config(os.environ[”MYAPP_CONFIG”]) application = myapp.create(config) To configure uWSGI, prepare a uwsgi.ini script as below − [uwsgi] master = 1 vacuum = true socket = 127.0.0.1:8080 enable-threads = true thunder-lock = true threads = 2 processes = 2 virtualenv = /path/to/venv wsgi-file = venv/src/wsgi.py chdir = venv/src uid = myapp-runner gid = myapp-runner You can now start the uWSGI like this − venv/bin/uwsgi -c uwsgi.ini Although uWSGI may serve HTTP requests directly, it can be helpful to use a reverse proxy such as NGINX. NGINX natively supports the uwsgi protocol, for efficiently proxying requests to uWSGI. Install Ngnix and then create an NGINX conf file that looks something like this − server { listen 80; server_name myproject.com; access_log /var/log/nginx/myproject-access.log; error_log /var/log/nginx/myproject-error.log warn; location / { uwsgi_pass 127.0.0.1:8080 include uwsgi_params; } } Finally start the Ngnix server. You should have a working application running. Print Page Previous Next Advertisements ”;

Python Falcon – Websocket

Python Falcon – Websocket ”; Previous Next A WebSocket is a persistent connection between a client and server to provide a bidirectional, full-duplex communication between the two. The communication takes place over HTTP through a single TCP/IP socket connection. It can be seen as an upgrade of HTTP instead of a protocol itself. One of the limitations of HTTP is that it is a strictly half-duplex or unidirectional protocol. With WebSockets, on the other hand, we can send message-based data, similar to UDP, but with the reliability of TCP. WebSocket uses HTTP as the initial transport mechanism, but keeps the TCP connection alive the connection after the HTTP response is received. Same connection object it can be used two-way communication between client and server. Thus, real-time applications can be built using WebSocket APIs. Falcon”s Websocket support is available only for ASGI applications. To provide Websocket capability, the resource class should have on_websocket() responder coroutine. async def on_websocket(self, req, ws): . . . Websocket requests can also be intercepted by hooks and middleware. Instead of the Response object, a falcon.asgi.WebSocket object is passed. How Does a WebSocket Function in Falcon? The following example demonstrates the functioning of WebSocket in Falcon application. First we have an on_get() responder that renders a template. Example The client browser displays a form with a text field and a button, When the button is clicked, a websocket object is created, and the on_websocket() responder is fired. It accepts the message input by the user and echoes it back to the client with a prefix “The message text was”. import falcon import falcon.asgi import jinja2 html = “”” <!DOCTYPE html> <html> <head> <title>Chat</title> </head> <body> <script> var ws = new WebSocket(“ws://localhost:8000/hello”); ws.onmessage = function(event) { var messages =document.getElementById(”messages”) var message = document.createElement(”li”) var content = document.createTextNode(event.data) message.appendChild(content) messages.appendChild(message) }; function sendMessage(event) { var input = document.getElementById(“messageText”) ws.send(input.value) input.value = ”” event.preventDefault() } </script> <h1>WebSocket Chat</h1> <form action=”” onsubmit=”sendMessage(event)”> <input type=”text” id=”messageText” autocomplete=”off”/> <button>Send</button> </form> <ul id=”messages”></ul> </body> </html> “”” class HelloResource: async def on_get(self, req, resp): “””Handles GET requests””” resp.status = falcon.HTTP_200 resp.content_type = ”text/html” template=jinja2.Template(html) resp.body=template.render() async def on_websocket(self, req, websocket): await websocket.accept() while True: data = await websocket.receive_text() await websocket.send_text(f”Message text was: {data}”) app = falcon.asgi.App() hello = HelloResource() app.add_route(”/hello”, hello) import uvicorn if __name__ == “__main__”: uvicorn.run(“main:app”, host=”0.0.0.0″, port=8000, reload=True) Output Start the Uvicorn server and visit http://localhost:8000/ws URL to display the chat form. Enter some text and press the Send button. Print Page Previous Next Advertisements ”;

Falcon – Suffixed Responders

Python Falcon – Suffixed Responders ”; Previous Next To understand the concept and the need of suffixed responders, let us define a StudentResource class. It consists of an on_get() responder that converts the students a list of dict objects to JSON and returns as its response. Let us also add on_post() responder that reads the data from the incoming request and adds a new dict object in the list. import falcon import json from waitress import serve students = [ {“id”: 1, “name”: “Ravi”, “percent”: 75.50}, {“id”: 2, “name”: “Mona”, “percent”: 80.00}, {“id”: 3, “name”: “Mathews”, “percent”: 65.25}, ] class StudentResource: def on_get(self, req, resp): resp.text = json.dumps(students) resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_JSON def on_post(self, req, resp): student = json.load(req.bounded_stream) students.append(student) resp.text = “Student added successfully.” resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_TEXT Using add_route() function of the Falcon”s App object, we add /students route. app = falcon.App() app.add_route(“/students”, StudentResource()) After starting the server, we can test the GET and POST requests from HTTPie command line − http GET localhost:8000/students HTTP/1.1 200 OK Content-Length: 187 Content-Type: application/json Date: Mon, 18 Apr 2022 06:21:02 GMT Server: waitress [ { “id”: 1, “name”: “Ravi”, “percent”: 75.5 }, { “id”: 2, “name”: “Mona”, “percent”: 80.0 }, { “id”: 3, “name”: “Mathews”, “percent”: 65.25 } ] http POST localhost:8000/students id=4 name=”Prachi” percent=59.90 HTTP/1.1 200 OK Content-Length: 27 Content-Type: text/plain; charset=utf-8 Date: Mon, 18 Apr 2022 06:20:51 GMT Server: waitress Student added successfully. Invoking on_get() again confirms the addition of new students resource. http GET localhost:8000/students HTTP/1.1 200 OK Content-Length: 187 Content-Type: application/json Date: Mon, 18 Apr 2022 06:21:02 GMT Server: waitress [ { “id”: 1, “name”: “Ravi”, “percent”: 75.5 }, { “id”: 2, “name”: “Mona”, “percent”: 80.0 }, { “id”: 3, “name”: “Mathews”, “percent”: 65.25 }, { “id”: “4”, “name”: “Prachi”, “percent”: “59.90” } ] At this stage, we would like to have a GET responder method in StudentResource class that reads the id parameter from the URL and retrieves a corresponding dict object of from the list. In other words, the URL of the format /student/{id} should be associated to the GET method in the resource class. But obviously, a class cannot have two methods of same name. Hence, we define to use suffix parameter for the add_route() method to distinguish between the two definitions of on_get() responders. A route with id parameter is added to the Application object by specifying suffix =”student”. app.add_route(“/students/{id:int}”, StudentResource(), suffix=”student”) We can now add another definition of on_get() method with this suffix, so that the name of this responder is on_get_student(), as follows − def on_get_student(self, req, resp, id): resp.text = json.dumps(students[id-1]) resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_JSON Start the Waitress server after adding the new route and on_get_student() responder and test this URL as follows − http GET localhost:8000/students/2 HTTP/1.1 200 OK Content-Length: 42 Content-Type: application/json Date: Mon, 18 Apr 2022 06:21:05 GMTy Server: waitress { “id”: 2, “name”: “Mona”, “percent”: 80.0 } Note that the on_put() responder (to update a resource) and on_delete() responder (to delete a resource) will also get invoked when the URL route /students/{id:int} is requested by the client with appropriate request header. We have already added this route with student as the suffix. Hence, on_put_student() method parses the path parameter in an integer variable. The JSON representation of the item with given id is fetched and updated with the data provided in the PUT request. def on_put_student(self, req, resp, id): student=students[id-1] data = json.load(req.bounded_stream) student.update(data) resp.text = json.dumps(student) resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_JSON The on_delete_student() responder simply deletes the item with the id specified in the DELETE request. The list of remaining resources is returned. def on_delete_student(self, req, resp, id): students.pop(id-1) resp.text = json.dumps(students) resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_JSON We can test the PUT and DELETE operations of the API with HTTPie commands − http PUT localhost:8000/students/2 id=3 name=”Mathews” percent=55 HTTP/1.1 200 OK Content-Length: 46 Content-Type: application/json Date: Sat, 18 Apr 2022 10:13:00 GMT Server: waitress { “id”: “3”, “name”: “Mathews”, “percent”: “55” } http DELETE localhost:8000/students/2 HTTP/1.1 200 OK Content-Length: 92 Content-Type: application/json Date: Sat, 18 Apr 2022 10:18:00 GMT Server: waitress [ { “id”: 1, “name”: “Ravi”, “percent”: 75.5 }, { “id”: 3, “name”: “Mathews”, “percent”: 65.25 } ] The complete code of this API (studentapi.py) is as under − import falcon import json from waitress import serve students = [ {“id”: 1, “name”: “Ravi”, “percent”: 75.50}, {“id”: 2, “name”: “Mona”, “percent”: 80.00}, {“id”: 3, “name”: “Mathews”, “percent”: 65.25}, ] class StudentResource: def on_get(self, req, resp): resp.text = json.dumps(students) resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_JSON def on_post(self, req, resp): student = json.load(req.bounded_stream) students.append(student) resp.text = “Student added successfully.” resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_TEXT def on_get_student(self, req, resp, id): resp.text = json.dumps(students[id-1]) resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_JSON def on_put_student(self, req, resp, id): student=students[id-1] data = json.load(req.bounded_stream) student.update(data) resp.text = json.dumps(student) resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_JSON def on_delete_student(self, req, resp, id): students.pop(id-1) print (students) resp.text = json.dumps(students) resp.status = falcon.HTTP_OK resp.content_type = falcon.MEDIA_JSON app = falcon.App() app.add_route(“/students”, StudentResource()) app.add_route(“/students/{id:int}”, StudentResource(), suffix=”student”) if __name__ == ”__main__”: serve(app, host=”0.0.0.0”, port=8000) Print Page Previous Next Advertisements ”;

Python Falcon – CORS

Python Falcon – CORS ”; Previous Next “Cross-Origin Resource Sharing” (CORS) is a situation when a frontend application that is running on one client browser tries to communicate with a backend through JavaScript code, and the backend is in a different “origin”than the frontend. The origin here is a combination of protocol, domain name and port numbers. As a result, http://localhost and https://localhost have different origins. If the browser with URL of one origin sends request for execution of JavaScript code from another origin, the browser sends OPTIONS http request. If the backend authorizes the communication from this different origin by sending the appropriate headers it will let the JavaScript in the frontend send its request to the backend. To enable the CORS policy for all responses, the Falcon app is configured as follows − from falcon import App app=App(cors_enable=True) To specify explicitly the allowed origins, import CORSMiddleware and add the list of origins to the app”s middleware, along with respective credentials. from falcon import App app = falcon.App(middleware=falcon.CORSMiddleware(allow_origins=”example.com”, allow_credentials=”*”) Print Page Previous Next Advertisements ”;

Python Falcon – Status Codes

Python Falcon – Status Codes ”; Previous Next By default, the HTTP server”s response to a client”s request is having 200 OK status. Falcon provides its own list of status constant for more convenience and readability. For example, 200 OK status code is represented by, resp.status = falcon.HTTP_OK These predefined Falcon constants avoid typos and cuts down on the number of string objects that must be created when preparing responses. However, starting with Falcon version 3.0, using bare int codes is allowed as well. resp.status = 200 For ASGI application, same status codes are applicable. Some of the status codes defined in Falcon library are listed below − Informational Codes HTTP_CONTINUE = HTTP_100 HTTP_SWITCHING_PROTOCOLS = HTTP_101 HTTP_PROCESSING = HTTP_102 Success Status Codes HTTP_OK = HTTP_200 HTTP_CREATED = HTTP_201 HTTP_ACCEPTED = HTTP_202 HTTP_NON_AUTHORITATIVE_INFORMATION = HTTP_203 HTTP_NO_CONTENT = HTTP_204 HTTP_RESET_CONTENT = HTTP_205 HTTP_PARTIAL_CONTENT = HTTP_206 HTTP_MULTI_STATUS = HTTP_207 HTTP_ALREADY_REPORTED = HTTP_208 HTTP_IM_USED = HTTP_226 Redirection Error Codes HTTP_MULTIPLE_CHOICES = HTTP_300 HTTP_MOVED_PERMANENTLY = HTTP_301 HTTP_FOUND = HTTP_302 HTTP_SEE_OTHER = HTTP_303 HTTP_NOT_MODIFIED = HTTP_304 HTTP_USE_PROXY = HTTP_305 HTTP_TEMPORARY_REDIRECT = HTTP_307 HTTP_PERMANENT_REDIRECT = HTTP_308 Client Error Codes HTTP_BAD_REQUEST = HTTP_400 HTTP_UNAUTHORIZED = HTTP_401 # “unauthenticated” HTTP_PAYMENT_REQUIRED = HTTP_402 HTTP_FORBIDDEN = HTTP_403 # “unauthorized” HTTP_NOT_FOUND = HTTP_404 HTTP_METHOD_NOT_ALLOWED = HTTP_405 HTTP_NOT_ACCEPTABLE = HTTP_406 HTTP_PROXY_AUTHENTICATION_REQUIRED = HTTP_407 HTTP_REQUEST_TIMEOUT = HTTP_408 HTTP_CONFLICT = HTTP_409 Server Error Codes HTTP_INTERNAL_SERVER_ERROR = HTTP_500 HTTP_NOT_IMPLEMENTED = HTTP_501 HTTP_BAD_GATEWAY = HTTP_502 HTTP_SERVICE_UNAVAILABLE = HTTP_503 HTTP_GATEWAY_TIMEOUT = HTTP_504 HTTP_HTTP_VERSION_NOT_SUPPORTED = HTTP_505 HTTP_INSUFFICIENT_STORAGE = HTTP_507 HTTP_LOOP_DETECTED = HTTP_508 HTTP_NETWORK_AUTHENTICATION_REQUIRED = HTTP_511 Print Page Previous Next Advertisements ”;

Python Falcon – Middleware

Python Falcon – Middleware ”; Previous Next A “middleware” is a function that is processed with every request (before being processed by any specific responder) as well as with every response before returning it. This function takes each request that comes to your application. A middleware works similar to hooks. However, unlike hooks, middleware methods apply globally to the entire App. It may perform some process with the request by running a code defined in it, and then passes the request to be processed the corresponding operation function. It can also process the response generated by the operation function before returning it. A middleware is a class that implements one or more of the following even handler methods. For a WSGI app, the methods are − process_request (self, req, resp) − This method processes the request before routing it. process_resource (self, req, resp, resource, params) − processes the request after routing. A dict object representing any additional params derived from the route”s URI template fields may be passed. process_response (self, req, resp, resource, req_succeeded) − This method is for post-processing of the response (after routing). The req_succeeded parameter is True if no exceptions were raised otherwise False. In case of the ASGI app, in addition to the above methods, the middleware class may define some more methods. To account for lifespan events, an optional part of ASGI specification, the startup and shutdown event handlers may be included. process_startup (self, scope, event) − This method processes the ASGI lifespan startup event. It is invoked when the server is ready to start up and receive connections, but before it has started to do so. process_shutdown(self, scope, event) − This method processes the ASGI lifespan shutdown event. It is invoked when the server has stopped accepting connections and closed all active connections. Since the ASGI application also responds to the requests under Websocket protocol, the middleware may define following coroutine methods − process_request_ws (self, req, ws) − This method processes a WebSocket handshake request before routing it. process_resource_ws (self, req, ws, resource, params) − This method processes a WebSocket handshake request after routing. A dict object derived from the route”s URI template fields may be passed to the resource”s responder. An instance of the middleware class has to be added to the Falcon application object at the time of initialization. For a WSGI Falcon app − class MyMiddleware: def process_request(self, req, resp): pass def process_resource(self, req, resp, resource, params): pass def process_response(self, req, resp, resource, req_succeeded): pass from falcon import App app=App(middleware=[MyMiddleware()]) For the ASGI app − class MyMiddleware: async def process_startup(self, scope, event): pass async def process_shutdown(self, scope, event): pass async def process_request(self, req, resp): pass async def process_resource(self, req, resp, resource, params): pass async def process_response(self, req, resp, resource, req_succeeded): pass async def process_request_ws(self, req, ws): pass async def process_resource_ws(self, req, ws, resource, params): pass from falcon.asgi import App app=App(middleware=[MyMiddleware()]) Print Page Previous Next Advertisements ”;

Python Falcon – Inspect Module

Python Falcon – Inspect Module ”; Previous Next The inspect module is a handy tool that provides information about registered routes and other components of a Falcon application such as middleware, sinks etc. The inspection of an application can be done by two ways – CLI tool and programmatically. The falcon-inspect-tool CLI script is executed from the command line giving the name of Python script in which Falcon application object is declared. For example, to inspect application object in studentapi.py − falcon-inspect-app studentapi:app Falcon App (WSGI) Routes: ⇒ /students – StudentResource: ├── GET – on_get └── POST – on_post ⇒ /students/{id:int} – StudentResource: ├── DELETE – on_delete_student ├── GET – on_get_student └── PUT – on_put_student The output shows registered routes and the responder methods in the resource class. To perform the inspection programmatically, use the application object as argument to inspect_app() function in the inspect module. from falcon import inspect from studentapi import app app_info = inspect.inspect_app(app) print(app_info) Save the above script as inspectapi.py and run it from the command line. python inspectapi.py Falcon App (WSGI) Routes: ⇒ /students – StudentResource: ├── GET – on_get └── POST – on_post ⇒ /students/{id:int} – StudentResource: ├── DELETE – on_delete_student ├── GET – on_get_student └── PUT – on_put_student Print Page Previous Next Advertisements ”;