- \n
- High performance<\/li>\n
- Automatic interactive code generation<\/li>\n
- Offers great editor support and documentation<\/li>\n<\/ul>\n
1.1 SQLite<\/h3>\n
SQLite<\/strong> is a software library that provides a relational database management system. It is lightweight in terms of setup, database administration, and required resources. It is self-contained, serverless, zero-configuration, transactional.<\/p>\n
- \n
- Self-contained means that it require minimal support from the operating system or any external library<\/li>\n
- Zero-configuration means that no external installation is required before using it<\/li>\n
- Transactional means it is fully ACID-compliant i.e. all queries and changes are atomic, consistent, isolated, and durable<\/li>\n<\/ul>\n
1.2 Setting up Python<\/h3>\n
If someone needs to go through the Python installation on Windows, please watch this<\/a> link. You can download the Python from this<\/a> link.<\/p>\n
2. CRUD Operations using Python FastAPI<\/h2>\n
I am using JetBrains PyCharm<\/a> as my preferred IDE. You’re free to choose the IDE of your choice. Fig. 1 represents the project structure for this tutorial.<\/p>\n
\n![\"crud](\"https:\/\/examples.javacodegeeks.com\/wp-content\/uploads\/2021\/12\/python-sqlite-crud-fastapi-projectstructure-img1.jpg\")
<\/a>Fig. 1: Application structure<\/figcaption><\/figure>\n<\/div>\n The file named –
movie_database.db<\/code> will be generated dynamically during the application run.<\/p>\n2.1 Creating a requirements file<\/h3>\n
Add the below code to the requirements file. The file will be responsible to download and install the packages required for this tutorial.<\/p>\n
requirements.txt<\/em><\/span><\/p>\n
fastapi\nsqlalchemy\npydantic\n<\/pre>\n
2.2 Creating the database config<\/h3>\n
Create the database configuration file. The file will be responsible for handling the SQLite database and interacting with the
movie<\/code> database.<\/p>\ndb_handler.py<\/em><\/span><\/p>\n
# Handling database\nfrom sqlalchemy import create_engine\nfrom sqlalchemy.ext.declarative import declarative_base\nfrom sqlalchemy.orm import sessionmaker\n\nDATABASE_URL = \"sqlite:\/\/\/.\/movie_database.db\"\n\n# creating engine\n# Setting check_same_thread to False so that the returned connection may be shared across multiple threads\nengine = create_engine(DATABASE_URL, connect_args={\"check_same_thread\": False})\n\n# bind \u2013 An optional Connectable, will assign the bind attribute on the MetaData instance\nSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)\n\n# A simple constructor that allows initialization\nBase = declarative_base()\n<\/pre>\n2.3 Creating the model class<\/h3>\n
Create the model class responsible for creating the table structure.
<\/div> <\/div><\/p>\nmodel.py<\/em><\/span><\/p>\n
# Model class\nfrom sqlalchemy import Boolean, Column, Integer, String\n\nfrom db_handler import Base\n\n\nclass Movies(Base):\n # Setting constraints on the table structure\n __tablename__ = \"movie\"\n id = Column(Integer, primary_key=True, autoincrement=True, index=True, nullable=False)\n movie_id = Column(String)\n movie_name = Column(String(255))\n director = Column(String(100))\n geners = Column(String)\n membership_required = Column(Boolean)\n cast = Column(String(255))\n streaming_platform = Column(String)\n<\/pre>\n
2.4 Creating the schema class<\/h3>\n
Create the schema class responsible for creating the schema and use it during the crud operations.<\/p>\n
schema.py<\/em><\/span><\/p>\n
# Schema class\nfrom typing import Optional\n\nfrom pydantic import BaseModel\n\n\nclass MovieBase(BaseModel):\n movie_name: str\n director: str\n geners: str\n cast: str\n\n\nclass MovieAdd(MovieBase):\n movie_id: int\n # Optional[str] is just a shorthand or alias for Union[str, None].\n streaming_platform: Optional[str] = None\n membership_required: bool\n\n # Behaviour of pydantic can be controlled via the Config class on a model\n # to support models that map to ORM objects. Config property orm_mode must be set to True\n class Config:\n orm_mode = True\n\n\nclass Movie(MovieAdd):\n id: int\n\n # Behaviour of pydantic can be controlled via the Config class on a model\n # to support models that map to ORM objects. Config property orm_mode must be set to True\n class Config:\n orm_mode = True\n\n\nclass UpdateMovie(BaseModel):\n # Optional[str] is just a shorthand or alias for Union[str, None].\n streaming_platform: Optional[str] = None\n membership_required: bool\n\n # Behaviour of pydantic can be controlled via the Config class on a model\n # to support models that map to ORM objects. Config property orm_mode must be set to True\n class Config:\n orm_mode = True\n<\/pre>\n
2.5 Creating the crud class<\/h3>\n
Create the crud class responsible for performing the sql operations.<\/p>\n
crud.py<\/em><\/span><\/p>\n
# Crud operation methods\nfrom sqlalchemy.orm import Session\n\nimport model\nimport schema\n\n\ndef get_all(db: Session):\n print(\"fetching records\")\n return db.query(model.Movies).all()\n\n\ndef get_by_id(db: Session, sl_id: int):\n print(\"fetching record id={}\".format(sl_id))\n return db.query(model.Movies).filter(model.Movies.id == sl_id).first()\n\n\ndef add(db: Session, movie: schema.MovieAdd):\n mv_details = model.Movies(\n movie_id=movie.movie_id,\n movie_name=movie.movie_name,\n director=movie.director,\n geners=movie.geners,\n membership_required=movie.membership_required,\n cast=movie.cast,\n streaming_platform=movie.streaming_platform)\n print(\"saving new record\")\n db.add(mv_details)\n db.commit()\n db.refresh(mv_details)\n return model.Movies(**movie.dict())\n\n\ndef delete(db: Session, sl_id: int):\n try:\n print(\"deleting record id={}\".format(sl_id))\n db.query(model.Movies).filter(model.Movies.id == sl_id).delete()\n db.commit()\n except Exception as e:\n raise Exception(e)\n\n\ndef update(db: Session, sl_id: int, details: schema.UpdateMovie):\n print(\"updating record id={}\".format(sl_id))\n db.query(model.Movies).filter(model.Movies.id == sl_id).update(vars(details))\n db.commit()\n return db.query(model.Movies).filter(model.Movies.id == sl_id).first()\n<\/pre>\n2.6 Creating the application class<\/h3>\n
Create the application controller responsible for handling the incoming requests from the user and interact with the database to show the results.<\/p>\n
application.py<\/em><\/span><\/p>\n
# Main class\nfrom typing import List\n\nfrom fastapi import Depends, FastAPI, HTTPException\nfrom sqlalchemy.orm import Session\n\nimport crud\nimport model\nimport schema\nfrom db_handler import SessionLocal, engine\n\nmodel.Base.metadata.create_all(bind=engine)\n\n# Initializing the app\napp = FastAPI(title=\"CRUD Operations using Python FastAPI\")\n\n\n# Database dependency\ndef get_db():\n db = SessionLocal()\n try:\n yield db\n finally:\n db.close()\n\n\n# Get all movies\n@app.get('\/get-all', response_model=List[schema.Movie])\ndef get_all(db: Session = Depends(get_db)):\n return crud.get_all(db=db)\n\n\n# Save new movie\n@app.post('\/add', response_model=schema.MovieAdd)\ndef add(movie: schema.MovieAdd, db: Session = Depends(get_db)):\n movie_id = crud.get_by_id(db=db, sl_id=movie.movie_id)\n if movie_id:\n print(\"Resource conflict\")\n raise HTTPException(status_code=409, detail=f\"Resource id {movie_id} already exist\")\n\n return crud.add(db=db, movie=movie)\n\n\n# Delete a movie by id\n@app.delete('\/delete')\ndef delete(sl_id: int, db: Session = Depends(get_db)):\n details = crud.get_by_id(db=db, sl_id=sl_id)\n if not details:\n print(\"Entity not found\")\n raise HTTPException(status_code=404, detail=f\"Resource not found\")\n try:\n crud.delete(db=db, sl_id=sl_id)\n except Exception as e:\n raise HTTPException(status_code=400, detail=f\"Unable to delete: {e}\")\n\n return {\"status\": \"accepted\", \"code\": \"202\", \"message\": \"Resource deleted\"}\n\n\n# Update a movie by id\n@app.put('\/update', response_model=schema.Movie)\ndef update(sl_id: int, update_param: schema.UpdateMovie, db: Session = Depends(get_db)):\n details = crud.get_by_id(db=db, sl_id=sl_id)\n if not details:\n print(\"Entity not found\")\n raise HTTPException(status_code=404, detail=f\"Resource not found\")\n\n return crud.update(db=db, details=update_param, sl_id=sl_id)\n\n\n# Driver code\nif __name__ == '__main__':\n import uvicorn\n\n uvicorn.run(app, host=\"localhost\", port=7001, log_level=\"debug\")\n<\/pre>\n3. Run the application<\/h2>\n
Run the
application.py<\/code> python script once the code is completed and if everything goes well the application will be started on the port number –7001<\/code> as shown in the below logs.<\/p>\nApplication logs<\/em><\/span><\/p>\n
INFO: Started server process [44004]\nINFO: Waiting for application startup.\nINFO: Application startup complete.\nINFO: Uvicorn running on http:\/\/localhost:7001 (Press CTRL+C to quit)\n<\/pre>\n
4. Demo<\/h2>\n
Open up the browser of your choice and hit the swagger documentation endpoint generated via the FastAPI. The documentation will list the endpoints created above.<\/p>\n
Documentation endpoint<\/em><\/span><\/p>\n
http:\/\/localhost:7001\/docs\n<\/pre>\n
If everything goes well the documentation page will be shown as in Fig. 1 and you can use the endpoints to set up a playground for the application.<\/p>\n
\n![\"crud](\"https:\/\/examples.javacodegeeks.com\/wp-content\/uploads\/2021\/12\/python-sqlite-crud-fastapi-projectdemo-img1.jpg\")
<\/a>Fig. 2: Swagger documentation<\/figcaption><\/figure>\n<\/div>\n That is all for this tutorial and I hope the article served you with whatever you were looking for. Happy Learning and do not forget to share!<\/p>\n
5. Summary<\/h2>\n
In this tutorial, we learned about the SQLite sqlalchemy crud operation and implemented the FastAPI framework to quickly set up our application. You can download the source code of this tutorial from the Downloads<\/a> section.<\/p>\n
<\/a>6. Download the Project<\/h2>\n
This was a tutorial on how to implement CRUD operations using the FastAPI framework.<\/p>\n