<jupyter_start><jupyter_text>Databases and ORM  Relational Database A key in the same table that is a primary key, if we reference this key on another table, then that is a foreign key. ![blog](https://raw.githubusercontent.com/azataiot/images/master/2021/12/22/blog.png)  NoSQL databases - key-value : Redis- graph : Neo4j- document oriented : MongoDB ![CleanShot2021-12-22at17.58.41@2x](https://raw.githubusercontent.com/azataiot/images/master/2021/12/22/CleanShot%202021-12-22%20at%2017.58.41@2x.png) one collection for users, another one for posts.  Communication with a SQL database with SQLAlchemy ![CleanShot2021-12-22at18.03.27](https://raw.githubusercontent.com/azataiot/images/master/2021/12/22/CleanShot%202021-12-22%20at%2018.03.27.png)<jupyter_code>!pip install 'databases[sqlite]'<jupyter_output>Requirement already satisfied: databases[sqlite] in /Users/azat/Library/Caches/pypoetry/virtualenvs/fastapi-data-science-FPk2pAXx-py3.9/lib/python3.9/site-packages (0.5.3)
Requirement already satisfied: sqlalchemy<1.5,>=1.4 in /Users/azat/Library/Caches/pypoetry/virtualenvs/fastapi-data-science-FPk2pAXx-py3.9/lib/python3.9/site-packages (from databases[sqlite]) (1.4.28)
Requirement already satisfied: aiosqlite in /Users/azat/Library/Caches/pypoetry/virtualenvs/fastapi-data-science-FPk2pAXx-py3.9/lib/python3.9/site-packages (from databases[sqlite]) (0.17.0)
Requirement already satisfied: greenlet!=0.4.17 in /Users/azat/Library/Caches/pypoetry/virtualenvs/fastapi-data-science-FPk2pAXx-py3.9/lib/python3.9/site-packages (from sqlalchemy<1.5,>=1.4->databases[sqlite]) (1.1.2)
Requirement already satisfied: typing_extensions>=3.7.2 in /Users/azat/Library/Caches/pypoetry/virtualenvs/fastapi-data-science-FPk2pAXx-py3.9/lib/python3.9/site-packages (from aiosqlite->databases[sqlite]) (4.0.1)
[...]<jupyter_text>![CGLn1w](https://raw.githubusercontent.com/azataiot/images/master/2021/12/22/CGLn1w.png)<jupyter_code># models.py

from datetime import datetime
from typing import Optional
import sqlalchemy
from pydantic import BaseModel, Field


class PostBase(BaseModel):
    title: str
    content: str
    publication_date: datetime = Field(default_factory=datetime.now)


class PostPatch(BaseModel):
    title: Optional[str] = None
    content: Optional[str] = None


class PostCreate(PostBase):
    pass


class PostDB(PostBase):
    id: int


""" metadata object: It's role is to keep all the information of a database schema together.
This is why you should create it only once in your whole project and always use the same one throughout.
"""
metadata = sqlalchemy.MetaData()

# we will define a table using the Table class
posts = sqlalchemy.Table(
    #  first argument is the name of the table, followed by the metadata object.
    "posts",
    metadata,
    # we list all the columns that should be defined in our table
    # name of the column, followed by its type and a certain number of options.
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, autoincrement=True),
    sqlalchemy.Column("publication_date", sqlalchemy.DateTime(), nullable=False),
    sqlalchemy.Column("title", sqlalchemy.String(length=255), nullable=False),
    sqlalchemy.Column("content", sqlalchemy.Text(), nullable=False)
)
metadata<jupyter_output><empty_output><jupyter_text>Connecting to a database<jupyter_code>from databases import Database

# database.py

# database engine, followed by authentication information and the hostname of the database server.
DATABASE_URL = "sqlite:///./db.data"
# we instantiate a Database instance using this URL.
# This is the connection layer provided by databases that will allow us to perform asynchronous queries.
database = Database(DATABASE_URL)
# We also define sqlalchemy_engine, which is the standard synchronous connection object provided by SQLAlchemy
sqlalchemy_engine = sqlalchemy.create_engine(DATABASE_URL)


# we define a simple function whose role is to simply return the database instance.

def get_database() -> Database:
    return database
# app.py
from typing import Optional, Tuple, List
from fastapi import FastAPI, status, Depends, Query, HTTPException

app = FastAPI()


# Decorating functions with the on_event decorators allows us to trigger some useful logic when FastAPI starts or stops.
@app.on_event("startup")
async def startup():
    # This will ensure that the database connection is open and ready to process requests.
    await get_database().connect()
    # The goal of this method is to create the table's schema inside our database.
    # This method is designed to work with a standard SQLAlchemy engine; this is why we instantiated it earlier.
    metadata.create_all(sqlalchemy_engine)


@app.on_event("shutdown")
async def shutdown():
    await get_database().disconnect()


async def get_pagination(
        skip: int = Query(0, ge=0),
        limit: int = Query(10, ge=10)
) -> Tuple[int, int]:
    capped_limit = min(100, limit)
    return skip, capped_limit


async def get_post_or_404(
        post_id: int, database: Database = Depends(get_database)
) -> PostDB:
    select_query = posts.select().where(posts.c.id == post_id)
    raw_post = await database.fetch_one(select_query)

    if raw_post is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
    return PostDB(**raw_post)


@app.get("/posts")
async def list_posts(
        pagination: Tuple[int, int] = Depends(get_pagination),
        database: Database = Depends(get_database),
) -> List[PostDB]:
    skip, limit = pagination
    select_query = posts.select().offset(skip).limit(limit)
    rows = await database.fetch_all(select_query)
    results = [PostDB(**row) for row in rows]
    return results


@app.get("/posts/{id}")
async def get_post(post: PostDB = Depends(get_post_or_404)) -> PostDB:
    return post


@app.post("/posts", response_model=PostDB, status_code=status.HTTP_201_CREATED)
async def create_post(
        post: PostCreate,
        database: Database = Depends(get_database),
) -> PostDB:
    # we rely on the SQLAlchemy expression language, which consists of chained method calls
    # This query is built directly from the posts object, which is the Table instance that we defined earlier.
    insert_query = posts.insert().values(post.dict())
    # Thanks to database, we can execute it asynchronously.
    post_id = await database.execute(insert_query)  #An INSERT query will return the id of the newly inserted row.
    post_db = await get_post_or_404(post_id, database)
    return post_db


@app.patch("/posts/{id}", response_model=PostDB)
async def update_post(
        post_update: PostPatch,
        post: PostDB = Depends(get_post_or_404),
        database: Database = Depends(get_database)
) -> PostDB:
    update_query = (
        posts.update().where(posts.id == id).values(post_update.dict(exclude_unset=True))
    )
    post_id = await database.execute(update_query)
    post_db = await get_post_or_404(post_id, database)
    return post_db


@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_post(post: PostDB = Depends(get_post_or_404), database: Database = Depends(get_database)):
    delete_query = posts.delete().where(posts.c.id == post.id)
    await database.execute(delete_query)<jupyter_output><empty_output><jupyter_text>Making Select Queries<jupyter_code>@app.get("/posts")
async def list_posts(
        pagination: Tuple[int, int] = Depends(get_pagination),
        database: Database = Depends(get_database),
) -> List[PostDB]:
    skip, limit = pagination
    select_query = posts.select().offset(skip).limit(limit)
    # we execute this query with the fetch_all method of database. This method will return a list of rows that match our query
    # Each row is returned to the form of a dictionary that associates column names and their values.
    rows = await database.fetch_all(select_query)

    results = [PostDB(**row) for row in rows]
    return results
# get a single object

@app.get("posts/{id}")
async def get_post(post: PostDB = Depends(get_post_or_404)) -> PostDB:
    return post
async def get_post_or_404(
        # you have to provide every argument manually since you are outside the dependency injection context.
        post_id: int, database: Database = Depends(get_database)
) -> PostDB:
    # The first part is to set the actual column we want to compare.
    # Each column is accessible via its name from the c attribute of the table object, that is, posts.c.id
    select_query = posts.select().where(posts.c.id == post_id)
    # Then, we simply call fetch_one on the database object.
    # It's a convenient shortcut when we only expect one row at most.
    raw_post = await database.fetch_one(select_query)

    if raw_post is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
    return PostDB(**raw_post)<jupyter_output><empty_output><jupyter_text>Making Update and delete Queries<jupyter_code>@app.patch("/posts/{id}", response_model=PostDB)
async def update_post(
        post_update: PostPatch,
        post: PostDB = Depends(get_post_or_404),
        database: Database = Depends(get_database)
) -> PostDB:
    update_query = (
        # we add a WHERE clause to only match the post we want to update.
        # since we are doing a partial update here,
        # you can see that we use the exclude_unset option to only get the values to update.
        posts.update().where(posts.id == id).values(post_update.dict(exclude_unset=True))
    )
    post_id = await database.execute(update_query)
    post_db = await get_post_or_404(post_id, database)
    return post_db<jupyter_output><empty_output><jupyter_text>Adding relationships relational databases are all about data and its relationships.<jupyter_code># models.py
import sqlalchemy as sql

comments = sql.Table(
    "comments",
    metadata,
    sql.Column("id", sql.Integer, primary_key=True, autoincrement=True),
    sql.Column("post_id", sql.ForeignKey("posts.id", ondelete="CASCADE"), nullable=False),
    sql.Column("publication_date", sql.DateTime(), nullable=False),
    sql.Column("content", sql.Text(), nullable=False),
)
# pydantic models of the Comments
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field


class CommentBase(BaseModel):
    post_id: int
    publication_date: datetime = Field(default_factory=datetime.now)
    content: str


class CommentCreate(CommentBase):
    pass


class CommentDB(CommentBase):
    id: int


class CommentUpdate(BaseModel):
    content: Optional[str] = None

from typing import List
from pydantic import BaseModel


class PostBase(BaseModel):
    title: str
    content: str
    publication_date: datetime = Field(default_factory=datetime.now)


class PostUpdate(BaseModel):
    title: Optional[str] = None
    content: Optional[str] = None


class PostCreate(PostBase):
    pass


class PostDB(PostBase):
    id: int


# Here, it'll be convenient to get the comments of a post in a single request.
# We'll use this model when getting a single post to serialize the comments along with the post data.
class PostPublic(PostBase):
    comments: List[CommentDB]
# create a new comment
from typing import cast, Mapping


@app.post("/comments", response_model=CommentDB, status_code=status.HTTP_201_CREATED)
async def create_comment(
        comment: CommentCreate,
        database: Database = Depends(get_database),
) -> CommentDB:
    # The only difference here is the first part of the function logic where
    # we check for the existence of the post before proceeding with the comment creation.
    select_post_query = posts.select().where(posts.c.id == comment.post_id)
    post = await database.fetch_one(select_post_query)

    if post is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Post {id} dose not exist."
        )

    insert_query = comments.insert().values(comment.dict())
    comment_id = database.execute(insert_query)

    select_query = comments.select().where(comments.c.id == comment_id)
    raw_coment = cast(Mapping, await database.fetch_one(select_query))
    return CommentDB(**raw_coment)
def get_post_or_404(
        id: int, database: Database = Depends(get_database),
) -> PostPublic:
    select_post_query = posts.select().where(posts.c.id == id)
    raw_post = await database.fetch_one(select_post_query)

    if raw_post is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND
        )

    # Here, you can see that we simply add a fetch_all query with the correct WHERE
    # statement to collect the comments associated with the post.
    select_post_comments_query = comments.select().where(comments.c.post_id == id)
    raw_comments = await database.fetch_all(select_post_comments_query)

    comments_list = [CommentDB(**comment) for comment in raw_comments]

    return PostPublic(**raw_post, comments=comments_list)<jupyter_output><empty_output><jupyter_text>Setting up a database migration systen with alembic<jupyter_code># going to include the app with a submodule<jupyter_output><empty_output><jupyter_text>Current updated structure of the app.``` bash.─ tree├── LICENSE├── README.md├── __init__.py├── alembic│   ├── README│   ├── env.py│   ├── script.py.mako│   └── versions├── alembic.ini├── app.py├── database.py├── models│   ├── __init__.py│   ├── comment.py│   └── post.py├── models.py├── poetry.lock└── pyproject.toml```It is somehow different from the one on the original app. ![5gQjWk](https://raw.githubusercontent.com/azataiot/images/master/2021/12/23/5gQjWk.png) I think it is a nice idea to create a models package in the project and place all models (including schematics) of one Class inside it's own python file.making that each of the .py file inside the modules package becomes an class like those in the java or the c alembic.ini file, which contains all the configuration options of Alembic.<jupyter_code># [alembic]
# # path to migration scripts
# script_location = alembic<jupyter_output><empty_output><jupyter_text>This setting expects the path of the alembic directory containing the migration files.In most of your projects, this will just be alembic, because it'll simply be at the root of your project. ![S0Lpmf](https://raw.githubusercontent.com/azataiot/images/master/2021/12/23/S0Lpmf.png)<jupyter_code># sqlalchemy.url = sqlite:///./app.db<jupyter_output><empty_output><jupyter_text>For alembic sqlalchemy.url i just placed the same for database.py DB_URL<jupyter_code># we also imported the metadata from the app.py to the env.py of for the alembic.

# import the metadata object.
# from database import metadata
# and then we changed the target to the metadata (that was None)<jupyter_output><empty_output><jupyter_text>``` add your model's MetaData object here for 'autogenerate' support from myapp import mymodel target_metadata = mymodel.Base.metadatatarget_metadata = metadata``` Now we will generate a new migration scriptgenerate a new migration script Made things work ![CleanShot2021-12-23at14.00.09@2x](https://raw.githubusercontent.com/azataiot/images/master/2021/12/23/CleanShot%202021-12-23%20at%2014.00.09@2x.png) After researching i got that, first we have to create the metadata object only once inside the database.py.then we use it everywhere like in the modules like in the app.pythen the two commands are helpful:- `alembic revision --autogenerate -m "Initial migration"`- `alembic upgrade head`and for submodules :- `git submodule update --recursive --remote`  Communicating with a SQL database with Tortoise ORM  Creating database models<jupyter_code># working with tortoise orm
from tortoise.models import Model
from tortoise import fields


class PostTortoise(Model):
    id = fields.IntField(pk=True, generated=True)
    publication_date = fields.DatetimeField(null=False)
    title = fields.CharField(max_length=255, null=False)
    content = fields.TextField(null=False)

    class Meta:
        table = "posts"
from pydantic import BaseModel


class PostBase(BaseModel):
    """Blog Post Base Class"""
    title: str
    content: str
    publication_date: datetime = Field(default_factory=datetime.now)

    class Config:
        orm_mode = True<jupyter_output><empty_output><jupyter_text>Setting up the Tortoise engine<jupyter_code>from tortoise.contrib.fastapi import register_tortoise
TORTOISE_ORM = {
    "connections": {"default": "sqlite://chapter6_tortoise.db"},
    "apps": {
        "models": {
            "models": ["chapter6.tortoise.models"],
            "default_connection": "default",
        },
    },
}

register_tortoise(
    app,
    config=TORTOISE_ORM,
    generate_schemas=True,
    add_exception_handlers=True,
)<jupyter_output><empty_output><jupyter_text>Connecting modules compatible with MongoDB<jupyter_code>@app.post("/posts", response_model=PostDB, status_code=status.HTTP_201_CREATED)
async def create_post(
    post: PostCreate, database: AsyncIOMotorDatabase = Depends(get_database)
) -> PostDB:
    post_db = PostDB(**post.dict())
    await database["posts"].insert_one(post_db.dict(by_alias=True))

    post_db = await get_post_or_404(post_db.id, database)

    return<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09947126837487653, 'fraction_numerical': 0.017605020045319855, 'mean_word_length': 4.467598475222363, 'pattern_counts': {'":': 6, '<': 54, '<?xml version=': 0, '>': 71, 'https://': 7, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '75271', 'license': 'MIT', 'n_tokens_mistral': 5541, 'n_tokens_neox': 5281, 'n_words': 1528, 'path': 'ch6-db-orm/db-orm.ipynb', 'repo_name': 'azataiot/building-datascience-applications-with-fast-api', 'size': '17211'}