Working with Databases in Python 3

Level: Intermediate

Table of Contents

1. Course Overview
2. Local relational database: SQLite

• Introduction to SQLite
• The sqlite3 module of the standard library
• Starter Code Review
• Using the sqlite3 module
• Demo — sqlite3 module
• Row Factories
• Demo — Row Factories
• Full code — Module 2
• Module 2 Summary

3. Module 3 — Relational database: PostgreSQL and psycopg2

• Introduction to PostgreSQL
• Installation of psycopg2
• Data insert
• Data Recovery
• Full demo
• Full code — Module 3
• Module 3 Summary

4. ORM: SQLAlchemy

• What is an ORM?
• Using SQLAlchemy
• Demonstration — Data Modeling
• Demo — CRUD Part 1 (SELECT, WHERE, GET)
• Demo — CRUD Part 2 (UPDATE, DELETE)
• Relationships (relationships between tables)
• Demonstration — Relationships Part 1
• Demonstration — Relationships Part 2
• Demo — Full CLI Application
• Database change
• Full code — Module 4
• Module 4 Summary

5. Local NoSQL database: Mongita

• Introduction to NoSQL Databases
• Presentation of Mongita
• Demonstration — Mongita
• Full code — Module 5
• Module 5 Summary

6. NoSQL database: MongoDB and PyMongo

• MongoDB in Visual Studio Code
• Migta to PyMongo migration
• Multi-condition filters
• Embedded Documents and Lists
• Demo — Embedded Documents and Lists
• Full code — Module 6
• Module 6 Summary

7. ODM: MongoEngine

• ODM vs ORM
• Data modeling with MongoEngine
• Demo — Modeling with MongoEngine
• Embedded Documents with MongoEngine
• Demo — Embedded Documents with MongoEngine
• Full code — Module 7
• Module 7 Summary

8. General summary and choice of the correct solution

1. Course Overview

This course, titled Working with Databases in Python 3, is taught by Douglas Starnes, independent author and speaker. It starts from a simple but fundamental observation: practically any Python application – or any other language – needs a database. However, the available databases are almost as diverse as the applications themselves.

Educational objectives

At the end of this course, you will know:

• Working with relational databases in Python (SQLite, PostgreSQL).
• Working with NoSQL databases in Python (Mongita, MongoDB).
• Use Python packages to access local file-based databases.
• Use low level packages with database servers.
• Use high level and more “Pythonic” packages (ORM, ODM).

Prerequisites

• Proficiency in Python 3.
• Basic knowledge of Visual Studio Code.

Tools used in the course

• Visual Studio Code (code editor, cross-platform and free).
• WSL (Windows Subsystem for Linux) — optional, used by the author to run Linux commands under Windows.
• VS Code Extensions: SQLite Explorer, PostgreSQL Explorer, MongoDB for VS Code.
• Docker to run PostgreSQL and MongoDB in containers.
• CoinGecko API (free, limited to ~10-30 requests/minute) to retrieve cryptocurrency prices in real time.

Common thread: the crypto portfolio management application

Throughout the course, a command line application for cryptocurrency portfolio management is gradually developed. It uses each database technology presented successively, which helps illustrate the differences and similarities between the approaches.

2. Local relational database: SQLite

2.1 Introduction to SQLite

SQLite is a relational database of a very particular kind: it does not require a server. Unlike PostgreSQL (which will be covered in the next module), SQLite stores the entire database in a local file. It is accessed via a command line client (sqlite3) or via a compatible API.

SQLite use cases

Note: SQLite is not designed for high-concurrency, multi-user applications. To go into production with many simultaneous users, we will prefer PostgreSQL or MySQL.

sqlite3 command line client

# Démarrer SQLite avec un fichier de base de données
sqlite3 portfolio.db

# Lister les tables
.tables

# Afficher le schéma d'une table
.schema investments

# Quitter
.quit

VS Code extension for SQLite

The author recommends installing a SQLite extension in VS Code to view tables and data directly in the editor, without using the terminal.

2.2 The standard library sqlite3 module

Python natively includes support for SQLite via the sqlite3 module — no third-party package installation required.

Basic workflow

import sqlite3

# 1. Connexion (crée le fichier si inexistant)
database = sqlite3.connect("portfolio.db")

# 2. Création d'un cursor
cursor = database.cursor()

# 3. Exécution de requêtes SQL
cursor.execute("CREATE TABLE IF NOT EXISTS investments (...);")

# 4. Commit pour persister les changements
database.commit()

Python Data Types ↔ SQLite

The sqlite3 module automatically manages the conversion between native Python types (bool, datetime) and their SQLite equivalents.

2.3 Boot Code Review

Before implementing SQLite support, the author introduces the third-party modules used in the demo:

• requests: to call the CoinGecko API and retrieve cryptocurrency prices. A simple GET request returns a JSON with the price.
• click: to easily create command line applications with commands, options and groups.

# Exemple d'appel à l'API CoinGecko
import requests

url = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd"
response = requests.get(url).json()
# Retourne : {"bitcoin": {"usd": 20000.0}}
price = response["bitcoin"]["usd"]

Structure of a click application with command group

import click

@click.group()
def cli():
    pass

@click.command()
@click.option("--coin_id", default="bitcoin")
@click.option("--currency", default="usd")
def show_coin_price(coin_id, currency):
    # ... logique
    pass

cli.add_command(show_coin_price)

if __name__ == "__main__":
    cli()

Using a command group (@click.group()) allows you to have several subcommands: python main.py show-coin-price, python main.py add-investment, etc. By default, click transforms underscores into hyphens in command names.

2.4 Using the sqlite3 module

Creating a table

import sqlite3

CREATE_INVESTMENTS_SQL = """
CREATE TABLE IF NOT EXISTS investments (
    coin_id TEXT,
    currency TEXT,
    amount REAL,
    sell INT,
    date TIMESTAMP
);
"""

database = sqlite3.connect("portfolio.db")
cursor = database.cursor()
cursor.execute(CREATE_INVESTMENTS_SQL)
database.commit()

Important: The IF NOT EXISTS clause avoids an error if the table already exists. This is a good practice during initialization.

Insertion with parameterized query

Insert uses placeholders (?) for values, which provides security against SQL injections and automatic Python type conversion.

import datetime

sql = "INSERT INTO investments VALUES (?, ?, ?, ?, ?);"
values = ("bitcoin", "usd", 1.0, False, datetime.datetime.now())
cursor.execute(sql, values)
database.commit()

Inserting multiple lines with executemany

rows = [
    ("bitcoin", "usd", 0.5, False, "2023-01-01 10:00:00"),
    ("ethereum", "usd", 5.0, True, "2023-02-15 14:30:00"),
]
sql = "INSERT INTO investments VALUES (?, ?, ?, ?, ?);"
cursor.executemany(sql, rows)
database.commit()

Data recovery

# Récupérer une seule ligne
result = cursor.execute("SELECT * FROM investments WHERE coin_id=?;", ("bitcoin",))
row = result.fetchone()
print(row)  # ('bitcoin', 'usd', 1.0, 0, '2023-01-01 10:00:00')

# Récupérer toutes les lignes
result = cursor.execute("SELECT * FROM investments;")
rows = result.fetchall()
for row in rows:
    print(row)

By default, results are tuples — values ​​are accessible by index (row[0], row[1], etc.).

2.5 Demo — sqlite3 module

The demo builds a complete command line application for managing a cryptocurrency wallet. The main features:

• show-coin-price: displays the current price of a cryptocurrency from CoinGecko.
• add-investment: adds an investment (purchase or sale) in the SQLite database.
• get-investment-value: calculates the current portfolio value for a given cryptocurrency.
• import-investments: import investments from a CSV file via executemany.

Import CSV with executemany

import csv

@click.command()
@click.option("--csv_file")
def import_investments(csv_file):
    with open(csv_file, "r") as f:
        rdr = csv.reader(f, delimiter=",")
        rows = list(rdr)
        sql = "INSERT INTO investments VALUES (?, ?, ?, ?, ?);"
        cursor.executemany(sql, rows)
        database.commit()
        print(f"Imported {len(rows)} investments from {csv_file}")

2.6 Row Factories

By default, sqlite3 returns data as lists of tuples. This can get confusing for complex tables, because you have to reference columns by their index rather than by their name. row factories allow tuples to be transformed into custom Python objects.

Dataclasses: the recommended solution

dataclasses (introduced in Python 3.7) eliminate boilerplate code from utility classes. They automatically generate:

• An __init__ based on type annotations.
• A __repr__ for display.
• Optional immutability.

from dataclasses import dataclass
import datetime

@dataclass
class Investment:
    coin_id: str
    currency: str
    amount: float
    sell: bool
    date: datetime.datetime

    def compute_value(self) -> float:
        return self.amount * get_coin_price(self.coin_id, self.currency)

With this dataclass, we can access values ​​by name: investment.coin_id, investment.amount, instead of row[0], row[2].

Comparison: tuple vs dataclass

Implementing a row factory

A row factory is a function that receives the cursor and raw tuple, and returns the desired object.

def investment_row_factory(_, row):
    return Investment(
        coin_id=row[0],
        currency=row[1],
        amount=row[2],
        sell=bool(row[3]),
        date=datetime.datetime.strptime(row[4], "%Y-%m-%d %H:%M:%S.%f")
    )

# Configuration de la row factory AVANT la création du cursor
database = sqlite3.connect("portfolio.db")
database.row_factory = investment_row_factory  # <-- AVANT cursor !
cursor = database.cursor()

Warning: The row_factory must be defined before obtaining the cursor. If set after, the results will always be tuples.

sqlite3 also includes a built-in row factory sqlite3.Row which transforms tuples into dictionary-like objects (access by column name).

2.7 Demo — Row Factories

The row factories demo is a copy of the code from the previous demo in row_factories.py, with the following changes:

1. Import of the dataclass decorator.
2. Added the Investment class with the dataclass.
3. Added investment_row_factory function.
4. Configuring database.row_factory in entry point.
5. Changed get_investment_value to use row.amount instead of row[0].

The advantage is not visible in a small demo, but in a real application with dozens of columns, accessing values ​​by name rather than index is much more maintainable. Additionally, type annotations allow VS Code to provide autocompletion.

2.8 Full code — Module 2

02/demos/main.py — Main application (without row factories)

import sqlite3
import datetime
import csv

import requests
import click

CREATE_INVESTMENTS_SQL = """
CREATE TABLE IF NOT EXISTS investments (
    coin_id TEXT,
    currency TEXT,
    amount REAL,
    sell INT, 
    date TIMESTAMP
);
"""

def get_coin_price(coin_id, currency):
    url = f"https://api.coingecko.com/api/v3/simple/price?ids={coin_id}&vs_currencies={currency}"
    data = requests.get(url).json()
    coin_price = data[coin_id][currency]
    return coin_price

@click.group()
def cli():
    pass

@click.command()
@click.option("--coin_id", default="bitcoin")
@click.option("--currency", default="usd")
def show_coin_price(coin_id, currency):
    coin_price = get_coin_price(coin_id, currency)
    print(f"The price of {coin_id} is {coin_price:.2f} {currency.upper()}")

@click.command()
@click.option("--coin_id")
@click.option("--currency")
@click.option("--amount", type=float)
@click.option("--sell", is_flag=True)
def add_investment(coin_id, currency, amount, sell):
    sql = "INSERT INTO investments VALUES (?, ?, ?, ?, ?);"
    values = (coin_id, currency, amount, sell, datetime.datetime.now())
    cursor.execute(sql, values)
    database.commit()

    if sell:
        print(f"Added sell of {amount} {coin_id}")
    else:
        print(f"Added buy of {amount} {coin_id}")

@click.command()
@click.option("--coin_id")
@click.option("--currency")
def get_investment_value(coin_id, currency):
    coin_price = get_coin_price(coin_id, currency)
    sql = """SELECT amount
    FROM investments
    WHERE coin_id=?
    AND currency=?
    AND sell=?;"""
    buy_result = cursor.execute(sql, (coin_id, currency, False)).fetchall()
    sell_result = cursor.execute(sql, (coin_id, currency, True)).fetchall()
    buy_amount = sum([row[0] for row in buy_result])
    sell_amount = sum([row[0] for row in sell_result])

    total = buy_amount - sell_amount

    print(f"You own a total of {total} {coin_id} worth {total * coin_price} {currency.upper()}")

@click.command()
@click.option("--csv_file")
def import_investments(csv_file):
    with open(csv_file, "r") as f:
        rdr = csv.reader(f, delimiter=",")
        rows = list(rdr)
        sql = "INSERT INTO investments VALUES (?, ?, ?, ?, ?);"
        cursor.executemany(sql, rows)
        database.commit()

        print(f"Imported {len(rows)} investments from {csv_file}")

cli.add_command(show_coin_price)
cli.add_command(add_investment)
cli.add_command(get_investment_value)
cli.add_command(import_investments)

if __name__ == "__main__":
    database = sqlite3.connect("portfolio.db")
    cursor = database.cursor()
    cursor.execute(CREATE_INVESTMENTS_SQL)
    cli()

02/demos/row_factories.py — Application with row factories and dataclasses

import sqlite3
import datetime
import csv
from dataclasses import dataclass

import requests
import click

CREATE_INVESTMENTS_SQL = """
CREATE TABLE IF NOT EXISTS investments (
    coin_id TEXT,
    currency TEXT,
    amount REAL,
    sell INT, 
    date TIMESTAMP
);
"""

@dataclass
class Investment:
    coin_id: str
    currency: str
    amount: float
    sell: bool
    date: datetime.datetime

    def compute_value(self) -> float:
        return self.amount * get_coin_price(self.coin_id, self.currency)

def investment_row_factory(_, row):
    return Investment(
        coin_id = row[0],
        currency = row[1],
        amount = row[2],
        sell = bool(row[3]),
        date = datetime.datetime.strptime(row[4], "%Y-%m-%d %H:%M:%S.%f")
    )

def get_coin_price(coin_id, currency):
    url = f"https://api.coingecko.com/api/v3/simple/price?ids={coin_id}&vs_currencies={currency}"
    data = requests.get(url).json()
    coin_price = data[coin_id][currency]
    return coin_price

@click.group()
def cli():
    pass

@click.command()
@click.option("--coin_id", default="bitcoin")
@click.option("--currency", default="usd")
def show_coin_price(coin_id, currency):
    coin_price = get_coin_price(coin_id, currency)
    print(f"The price of {coin_id} is {coin_price:.2f} {currency.upper()}")

@click.command()
@click.option("--coin_id")
@click.option("--currency")
@click.option("--amount", type=float)
@click.option("--sell", is_flag=True)
def add_investment(coin_id, currency, amount, sell):
    sql = "INSERT INTO investments VALUES (?, ?, ?, ?, ?);"
    values = (coin_id, currency, amount, sell, datetime.datetime.now())
    cursor.execute(sql, values)
    database.commit()

    if sell:
        print(f"Added sell of {amount} {coin_id}")
    else:
        print(f"Added buy of {amount} {coin_id}")

@click.command()
@click.option("--coin_id")
@click.option("--currency")
def get_investment_value(coin_id, currency):
    coin_price = get_coin_price(coin_id, currency)
    sql = """SELECT *
    FROM investments
    WHERE coin_id=?
    AND currency=?
    AND sell=?;"""
    buy_result = cursor.execute(sql, (coin_id, currency, False)).fetchall()
    sell_result = cursor.execute(sql, (coin_id, currency, True)).fetchall()
    buy_amount = sum([row.amount for row in buy_result])
    sell_amount = sum([row.amount for row in sell_result])

    total = buy_amount - sell_amount

    print(f"You own a total of {total} {coin_id} worth {total * coin_price} {currency.upper()}")

@click.command()
@click.option("--csv_file")
def import_investments(csv_file):
    with open(csv_file, "r") as f:
        rdr = csv.reader(f, delimiter=",")
        rows = list(rdr)
        sql = "INSERT INTO investments VALUES (?, ?, ?, ?, ?);"
        cursor.executemany(sql, rows)
        database.commit()

        print(f"Imported {len(rows)} investments from {csv_file}")

cli.add_command(show_coin_price)
cli.add_command(add_investment)
cli.add_command(get_investment_value)
cli.add_command(import_investments)

if __name__ == "__main__":
    database = sqlite3.connect("portfolio.db")
    database.row_factory = investment_row_factory
    cursor = database.cursor()
    cursor.execute(CREATE_INVESTMENTS_SQL)
    cli()

02/demos/requirements.txt

requests
click

2.9 Module 2 Summary

• SQLite is a file-based, serverless database.
• It is accessible via the sqlite3 command line client or the VS Code extension.
• The sqlite3 module of the Python standard library allows you to connect, execute SQL queries and retrieve the results.
• Parameterized queries (with ?) ensure automatic conversion of Python types to SQLite and protect against SQL injections.
• executemany allows you to insert multiple lines in a single operation.
• By default, the results are lists of tuples, but row factories allow them to be transformed into custom Python objects.
• dataclasses (Python 3.7+) are recommended for creating simple and readable model classes.

3. Relational database: PostgreSQL and psycopg2

3.1 Introduction to PostgreSQL

PostgreSQL is among the most popular relational databases in the Python ecosystem. Unlike SQLite, it is a database server: the connection is via the network, and authentication (username and password) is required.

The Python package for communicating with PostgreSQL is psycopg2. The working model is similar to that of SQLite:

1. Establish a connection.
2. Send SQL statements.
3. Retrieve and parse data.

The main advantage of psycopg2 over sqlite3 is access to PostgreSQL-specific features (advanced types, execute_values, cursors factories, etc.).

3.2 Installing psycopg2

There are two versions of the package:

pip install psycopg2-binary

Connecting to PostgreSQL

import psycopg2

connection = psycopg2.connect(
    host="localhost",
    database="manager",
    user="postgres",
    password="pgpassword"
)

cursor = connection.cursor()

VS Code extension for PostgreSQL

The author recommends the Chris Kolkman extension (elephant icon in sidebar) to visualize PostgreSQL tables and data directly in VS Code. The elephant is the official mascot of PostgreSQL.

To connect: click on the PostgreSQL icon → click on the + → enter host, user, password and port (5432 by default).

3.3 Data insertion

Simple insertion (with risk of SQL injection — avoid in production)

The demo example uses an f-string to construct the query, which is only acceptable here because the data comes directly from the user via click which normalizes it. In production, you should always use parameterized queries.

stmt = f"""
    insert into investment (
        coin, currency, amount
    ) values (
        '{coin.lower()}', '{currency.lower()}', {amount}
    )
"""

connection = get_connection()
cursor = connection.cursor()
cursor.execute(stmt)
connection.commit()
cursor.close()
connection.close()

Batch insert with execute_values

psycopg2.extras.execute_values allows inserting multiple rows very efficiently, without manual looping.

import psycopg2.extras

stmt = "insert into investment (coin, currency, amount) values %s"
rows = [["bitcoin", "usd", 1.0], ["ethereum", "eur", 5.0]]

connection = get_connection()
cursor = connection.cursor()
psycopg2.extras.execute_values(cursor, stmt, rows)
connection.commit()
cursor.close()
connection.close()

3.4 Data Recovery

Basic recovery (results in tuples)

cursor.execute("SELECT * FROM investment WHERE coin='bitcoin';")
row = cursor.fetchone()   # Une seule ligne
# ou
rows = cursor.fetchall()  # Toutes les lignes

Cursor factory: RealDictCursor

To obtain results similar to Python dictionaries, psycopg2 offers the RealDictCursor in the psycopg2.extras module.

cursor = connection.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cursor.execute("SELECT * FROM investment;")
rows = cursor.fetchall()
# rows est une liste de RealDictRow (chaque ligne est un objet dict-like)

Conversion to dataclasses

Exactly as with sqlite3, we can convert RealDictRow into dataclasses:

from dataclasses import dataclass

@dataclass
class Investment:
    id: int
    coin: str
    currency: str
    amount: float

cursor = connection.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cursor.execute("SELECT * FROM investment;")
data = [Investment(**dict(row)) for row in cursor.fetchall()]

3.5 Full Demo

The demo implements three click commands:

• new-investment: adds an investment to PostgreSQL with interactive prompt.
• import-investments: import from a CSV using execute_values.
• view-investments: displays all investments with their current value.

A get_connection() utility function centralizes connection logic to avoid code duplication.

3.6 Full code — Module 3

03/demos/manager.py

from dataclasses import dataclass
import csv

import click
import requests
import psycopg2
import psycopg2.extras

@dataclass
class Investment:
    id: int
    coin: str
    currency: str
    amount: float

def get_connection():
    connection = psycopg2.connect(
        host="localhost",
        database="manager",
        user="postgres",
        password="pgpassword"
    )
    return connection

@click.group()
def cli():
    pass

@click.command()
@click.option("--coin", prompt=True)
@click.option("--currency", prompt=True)
@click.option("--amount", prompt=True)
def new_investment(coin, currency, amount):
    stmt = f"""
        insert into investment (
            coin, currency, amount
        ) values (
            '{coin.lower()}', '{currency.lower()}', {amount}
        )
    """

    connection = get_connection()
    cursor = connection.cursor()

    cursor.execute(stmt)
    connection.commit()

    cursor.close()
    connection.close()
    
    print(f"Added investment for {amount} {coin} in {currency}")

@click.command()
@click.option("--filename")
def import_investments(filename):
    stmt = "insert into investment (coin, currency, amount) values %s"

    connection = get_connection()
    cursor = connection.cursor()
    
    with open(filename, 'r') as f:
        coin_reader = csv.reader(f)
        rows = [[x.lower() for x in row[1:]] for row in coin_reader]

    psycopg2.extras.execute_values(cursor, stmt, rows)
    connection.commit()

    cursor.close()
    connection.close()

    print(f"Added {len(rows)} investments")

@click.command()
@click.option("--currency")
def view_investments(currency):
    connection = get_connection()
    cursor = connection.cursor(cursor_factory=psycopg2.extras.RealDictCursor)

    stmt = "select * from investment"

    if currency is not None:
        stmt += f" where currency='{currency.lower()}'"

    cursor.execute(stmt)
    data = [Investment(**dict(row)) for row in cursor.fetchall()]

    cursor.close()
    connection.close()

    coins = set([row.coin for row in data])
    currencies = set([row.currency for row in data])

    url = f"https://api.coingecko.com/api/v3/simple/price?ids={','.join(coins)}&vs_currencies={','.join(currencies)}"
    coin_data = requests.get(url).json()

    for investment in data:
        coin_price = coin_data[investment.coin][investment.currency.lower()]
        coin_total = investment.amount * coin_price
        print(f"{investment.amount} {investment.coin} in {investment.currency} is worth {coin_total}")

cli.add_command(new_investment)
cli.add_command(import_investments)
cli.add_command(view_investments)

if __name__ == "__main__":
    cli()

03/demos/requirements.txt

psycopg2-binary
click
requests

3.7 Module 3 Summary

• PostgreSQL is a database server — network connection + authentication required.
• The psycopg2 package (or psycopg2-binary for development) is the standard Python driver for PostgreSQL.
• Working model is same as SQLite: connection → cursor → SQL → results.
• psycopg2.extras.execute_values allows batch insert without looping.
• The RealDictCursor returns results similar to dictionaries, convertible to dataclasses.
• In practice, we almost never use psycopg2 directly — we prefer an ORM like SQLAlchemy, which reduces or even eliminates SQL.

4. ORM: SQLAlchemy

4.1 What is an ORM?

An ORM (Object Relational Mapper) is a tool allowing you to manipulate a relational database using a general language (Python), without writing SQL. It creates a correspondence (mapping) between Python classes and database tables.

Problem resolved by an ORM

With sqlite3 and psycopg2, Python is mainly used to “transport” SQL to the server and wait for the data in return. The real work is done in SQL. An ORM offers several advantages:

4.2 Using SQLAlchemy

SQLAlchemy is distributed as a Python package:

pip install sqlalchemy

SQLAlchemy offers two APIs:

The course uses both together.

Creating a data model

from sqlalchemy import String, Numeric, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

# Toutes les classes de modèle doivent hériter d'une sous-classe de DeclarativeBase
class Base(DeclarativeBase):
    pass

class Investment(Base):
    __tablename__ = "investment"  # Nom de la table en base

    id: Mapped[int] = mapped_column(primary_key=True)
    coin: Mapped[str] = mapped_column(String(32))
    currency: Mapped[str] = mapped_column(String(3))
    amount: Mapped[float] = mapped_column(Numeric(5, 2))

    def __repr__(self) -> str:
        return f"<Investment coin: {self.coin}, currency: {self.currency}, amount: {self.amount}>"

Key points:

• DeclarativeBase is the base class for all models. We create a Base subclass (even empty) and inherit from it.
• Mapped[type] configures the Python side of the mapping.
• mapped_column() configures the database side.
• Python type annotations are used to define schema — this is idiomatic to modern Python.

Connection and creation of tables

# SQLite
engine = create_engine("sqlite:///manager.db")

# PostgreSQL
engine = create_engine("postgresql://postgres:pgpassword@localhost/manager")

# Créer toutes les tables définies dans les classes héritant de Base
Base.metadata.create_all(engine)

The connection string is the only place to change to change databases.

Session

The Session is the main interface for CRUD operations.

from sqlalchemy.orm import Session

with Session(engine) as session:
    # ... opérations CRUD
    session.commit()

4.3 Demonstration — Data Modeling

from sqlalchemy import String, Numeric, Text, ForeignKey, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship
from typing import List

class Base(DeclarativeBase):
    pass

class Investment(Base):
    __tablename__ = "investment"

    id: Mapped[int] = mapped_column(primary_key=True)
    coin: Mapped[str] = mapped_column(String(32))
    currency: Mapped[str] = mapped_column(String(3))
    amount: Mapped[float] = mapped_column(Numeric(5, 2))

    def __repr__(self) -> str:
        return f"<Investment coin: {self.coin}, currency: {self.currency}, amount: {self.amount}>"

engine = create_engine("sqlite:///demo.db")
Base.metadata.create_all(engine)

Running this file creates the demo.db file and the investment table with the defined columns. We can check with SQLite explorer in VS Code.

4.4 Demonstration — CRUD Part 1 (SELECT, WHERE, GET)

Insertion

from sqlalchemy.orm import Session

investments = [
    Investment(coin="bitcoin", currency="usd", amount=1.0),
    Investment(coin="ethereum", currency="usd", amount=10.0),
    Investment(coin="dogecoin", currency="usd", amount=1000.0),
]

with Session(engine) as session:
    session.add_all(investments)
    session.commit()

SELECT with the Core API

from sqlalchemy import select

# SELECT * FROM investment
stmt = select(Investment)

# Afficher le SQL généré
print(stmt)  # SELECT investment.id, investment.coin, ... FROM investment

with Session(engine) as session:
    results = session.execute(stmt).scalars().all()
    for inv in results:
        print(inv)

WHERE — filtering

# SELECT * FROM investment WHERE investment.coin = 'bitcoin'
stmt = select(Investment).where(Investment.coin == "bitcoin")

with Session(engine) as session:
    bitcoin = session.execute(stmt).scalar_one()
    print(bitcoin)

GET by primary key

with Session(engine) as session:
    investment = session.get(Investment, 1)  # id = 1
    print(investment)

4.5 Demo — CRUD Part 2 (UPDATE, DELETE)

UPDATE

with Session(engine) as session:
    bitcoin = session.get(Investment, 1)
    
    # Modifier l'objet — il est automatiquement marqué "dirty"
    bitcoin.amount = 1.234
    
    # Les objets modifiés mais non encore persistés
    print(session.dirty)  # IdentitySet({<Investment coin: bitcoin ...>})
    
    session.commit()  # Persiste les changements

SQLAlchemy automatically detects modified objects in the session (dirty state) and generates the corresponding SQL UPDATE during commit.

DELETE

with Session(engine) as session:
    dogecoin = session.get(Investment, 3)
    
    session.delete(dogecoin)
    
    # Objets marqués pour suppression
    print(session.deleted)
    
    session.commit()  # Exécute le DELETE SQL

4.6 Relationships (relationships between tables)

relationships allow you to navigate between linked tables directly from Python objects.

Modeling with ForeignKey

class Portfolio(Base):
    __tablename__ = "portfolio"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(256))
    description: Mapped[str] = mapped_column(Text())

    investments: Mapped[List["Investment"]] = relationship(back_populates="portfolio")

    def __repr__(self) -> str:
        return f"<Portfolio name: {self.name} with {len(self.investments)} investment(s)>"

class Investment(Base):
    __tablename__ = "investment"

    id: Mapped[int] = mapped_column(primary_key=True)
    coin: Mapped[str] = mapped_column(String(32))
    currency: Mapped[str] = mapped_column(String(3))
    amount: Mapped[float] = mapped_column(Numeric(5, 2))

    portfolio_id: Mapped[int] = mapped_column(ForeignKey("portfolio.id"))
    portfolio: Mapped["Portfolio"] = relationship(back_populates="investments")

    def __repr__(self) -> str:
        return f"<Investment coin: {self.coin}, currency: {self.currency}, amount: {self.amount}>"

Important points:

• ForeignKey("portfolio.id"): refers to the table name (lowercase), not the class name.
• relationship(): Python convenience attribute for navigating to the related object — no physical columns created in the table.
• back_populates: configures the bidirectional relationship (you can go from Investment to Portfolio and vice-versa).

4.7 Demonstration — Relationships Part 1

# Création d'un portfolio et de ses investissements
with Session(engine) as session:
    portfolio = Portfolio(
        name="My Portfolio",
        description="Main cryptocurrency portfolio"
    )
    
    investments = [
        Investment(coin="bitcoin", currency="usd", amount=1.0),
        Investment(coin="ethereum", currency="usd", amount=5.0),
    ]
    
    # Associer les investissements au portfolio
    for inv in investments:
        portfolio.investments.append(inv)
    
    session.add(portfolio)
    session.commit()

SQLAlchemy automatically manages the update of the portfolio_id in the investment table.

4.8 Demonstration — Relationships Part 2

Join with the Core API

from sqlalchemy import select

# Jointure simple (tous les investissements de chaque portfolio)
stmt = select(Portfolio, Investment).join(Investment.portfolio)
print(stmt)  # Affiche le SQL généré — complexe !

# Jointure filtrée : portfolios avec des investissements en bitcoin
stmt = (
    select(Portfolio)
    .join(Portfolio.investments)
    .where(Investment.coin == "bitcoin")
)

with Session(engine) as session:
    portfolios_with_bitcoin = session.execute(stmt).scalars().all()
    for p in portfolios_with_bitcoin:
        print(p)

The SQLAlchemy Core API generates the join SQL automatically. This is much simpler and less error prone than writing join SQL manually.

4.9 Demo — Complete CLI Application

The final application manager.py implements four commands:

• add-portfolio: creates a new portfolio.
• add-investment: creates an investment and associates it with an existing portfolio.
• view-portfolio: displays investments in a portfolio with their current value.
• clear-database: clears all tables (useful for development).

@click.command(help="View the investments in a portfolio")
def view_portfolio():
    with Session(engine) as session:
        stmt = select(Portfolio)
        all_portfolios = session.execute(stmt).scalars().all()

        for index, portfolio in enumerate(all_portfolios):
            print(f"{index + 1}: {portfolio.name}")

        portfolio_id = int(input("Select a portfolio: ")) - 1
        portfolio = all_portfolios[portfolio_id]

        investments = portfolio.investments

        coins = set([investment.coin for investment in investments])
        currencies = set([investment.currency for investment in investments])

        coin_prices = get_coin_prices(coins, currencies)

        print(f"Investments in {portfolio.name}")
        for index, investment in enumerate(investments):
            coin_price = coin_prices[investment.coin][investment.currency.lower()]
            total_price = float(investment.amount) * coin_price
            print(f"{index + 1}: {investment.coin} {total_price:.2f} {investment.currency}")

        print("Prices provided by CoinGecko")

4.10 Database change

This is the kept promise of SQLAlchemy: only one line of code to change to move from SQLite to PostgreSQL.

# Développement (SQLite — aucune installation serveur requise)
engine = create_engine("sqlite:///manager.db")

# Production (PostgreSQL — remplacer cette ligne uniquement)
engine = create_engine("postgresql://postgres:pgpassword@localhost/manager")

Note: SQLAlchemy can create the tables for you (create_all), but cannot create the PostgreSQL database itself. You must create the empty database manually (eg: CREATE DATABASE manager; in psql).

SQLAlchemy automatically adapts data types between SQLite and PostgreSQL. For example, a Numeric(5, 2) will have a slightly different representation in each database, but SQLAlchemy maintains consistent behavior on the Python side.

4.11 Full code — Module 4

04/demos/manager.py

from typing import List

from sqlalchemy import String, Numeric, create_engine, select, Text, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship

import click
import requests

def get_coin_prices(coins, currencies):
    coin_csv = ",".join(coins)
    currency_csv = ",".join(currencies)

    COINGECKO_URL = f"https://api.coingecko.com/api/v3/simple/price?ids={coin_csv}&vs_currencies={currency_csv}"

    data = requests.get(COINGECKO_URL).json()

    return data

class Base(DeclarativeBase):
    pass

class Portfolio(Base):
    __tablename__ = "portfolio"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(256))
    description: Mapped[str] = mapped_column(Text())

    investments: Mapped[List["Investment"]] = relationship(
        back_populates="portfolio")

    def __repr__(self) -> str:
        return f"<Portfolio name: {self.name} with {len(self.investments)} investment(s)>"

class Investment(Base):
    __tablename__ = "investment"

    id: Mapped[int] = mapped_column(primary_key=True)
    coin: Mapped[str] = mapped_column(String(32))
    currency: Mapped[str] = mapped_column(String(3))
    amount: Mapped[float] = mapped_column(Numeric(5, 2))

    portfolio_id: Mapped[int] = mapped_column(ForeignKey("portfolio.id"))
    portfolio: Mapped["Portfolio"] = relationship(
        back_populates="investments")

    def __repr__(self) -> str:
        return f"<Investment coin: {self.coin}, currency: {self.currency}, amount: {self.amount}>"

# engine = create_engine("sqlite:///manager.db")
engine = create_engine("postgresql://postgres:pgpassword@localhost/manager")
Base.metadata.create_all(engine)

@click.group()
def cli():
    pass

@click.command(help="View the investments in a portfolio")
def view_portfolio():
    with Session(engine) as session:
        stmt = select(Portfolio)
        all_portfolios = session.execute(stmt).scalars().all()

        for index, portfolio in enumerate(all_portfolios):
            print(f"{index + 1}: {portfolio.name}")

        portfolio_id = int(input("Select a portfolio: ")) - 1
        portfolio = all_portfolios[portfolio_id]

        investments = portfolio.investments

        coins = set([investment.coin for investment in investments])
        currencies = set([investment.currency for investment in investments])

        coin_prices = get_coin_prices(coins, currencies)

        print(f"Investments in {portfolio.name}")
        for index, investment in enumerate(investments):
            coin_price = coin_prices[investment.coin][investment.currency.lower()]
            total_price = float(investment.amount) * coin_price
            print(
                f"{index + 1}: {investment.coin} {total_price:.2f} {investment.currency}")

        print("Prices provided by CoinGecko")

@click.command(help="Create a new investment and add it to a portfolio")
@click.option("--coin", prompt=True)
@click.option("--currency", prompt=True)
@click.option("--amount", prompt=True)
def add_investment(coin, currency, amount):
    with Session(engine) as session:
        stmt = select(Portfolio)
        all_portfolios = session.execute(stmt).scalars().all()

        for index, portfolio in enumerate(all_portfolios):
            print(f"{index + 1}: {portfolio.name}")

        portfolio_index = int(input("Select a portfolio: ")) - 1
        portfolio = all_portfolios[portfolio_index]

        investment = Investment(coin=coin, currency=currency, amount=amount)
        portfolio.investments.append(investment)

        session.add(portfolio)
        session.commit()

        print(f"Added new {coin} investment to {portfolio.name}")

@click.command(help="Create a new portfolio")
@click.option("--name", prompt=True)
@click.option("--description", prompt=True)
def add_portfolio(name, description):
    portfolio = Portfolio(name=name, description=description)
    with Session(engine) as session:
        session.add(portfolio)
        session.commit()
    print(f"Added portfolio {name}")

@click.command(help="Drop all tables in the database")
def clear_database():
    Base.metadata.drop_all(engine)
    print("Database cleared!")

cli.add_command(clear_database)
cli.add_command(add_portfolio)
cli.add_command(add_investment)
cli.add_command(view_portfolio)

if __name__ == "__main__":
    cli()

04/demos/requirements.txt

sqlalchemy
psycopg2-binary
click
requests

4.12 Summary of Module 4

• An ORM (Object Relational Mapper) allows you to manipulate a database in Python without writing SQL.
• SQLAlchemy offers two APIs: the ORM for modeling, and the Core API for CRUD queries.
• Model classes inherit from a subclass of DeclarativeBase.
• Python type annotations (Mapped[str], Mapped[int]) define mappings.
• create_engine() with a connection string is the only configuration point related to database type.
• Base.metadata.create_all(engine) automatically creates the tables.
• The Session is the central object for CRUD operations.
• relationships allow you to navigate between linked tables directly from Python objects.
• Switching from SQLite to PostgreSQL only requires a change in the connection string.

5. Local NoSQL database: Mongita

5.1 Introduction to NoSQL Databases

A common misconception: NoSQL databases would replace SQL databases. This is false. NoSQL databases complement relational databases, they do not compete with them.

What is a NoSQL database?

NoSQL basically means “anything that is not relational”. The only commonality between NoSQL databases is that they do not store data in tabular tables linked by common keys.

Types of NoSQL databases

Redis, thanks to its in-memory storage, can support millions of operations per second — ideal for caching.

When to use NoSQL?

• Data whose schema evolves frequently (no table migrations).
• Natural JSON documents (logs, events, configs).
• Large horizontally scalable volumes.
• Case where schema flexibility takes precedence over referential integrity.

5.2 Mongita Overview

Mongita is the equivalent of SQLite for document-oriented databases: a local, serverless database, stored in files.

Key point: Mongita implements a subset of the PyMongo API. Everything you learn with Mongita applies directly to MongoDB with PyMongo.

Installation

pip install mongita

Connection and access to collections

from mongita import MongitaClientDisk

# Crée un client qui stocke les données sur le disque
client = MongitaClientDisk()

# Les bases de données et collections sont créées dynamiquement au premier accès
db = client.portfolio
investments = db.investments

Basic operations

# INSERT — insertion d'un document (dictionnaire Python)
investment_document = {
    "coin_id": "bitcoin",
    "currency": "usd",
    "amount": 1.0,
    "sell": False,
    "timestamp": "2023/01/01 10:00:00"
}
investments.insert_one(investment_document)

# FIND — récupération de documents avec un filtre
result = investments.find({"coin_id": "bitcoin", "sell": False})
for doc in result:
    print(doc)  # Dictionnaire Python

# UPDATE — mise à jour avec un opérateur $inc
filter_doc = {"coin_id": "bitcoin"}
update_doc = {"$inc": {"amount": 0.5}}
investments.update_one(filter_doc, update_doc)

# DELETE
investments.delete_one({"coin_id": "dogecoin"})

Filter documents and update documents

• Filter document: specifies the selection criteria (similar to the WHERE clause).
• Update document: specifies the actions to perform ($inc, $set, $push, $pull).

5.3 Demonstration — Mongita

The demo starts from the SQLite demo code (module 2) and replaces all parts specific to sqlite3 with Mongita code.

Key changes:

1. Delete the sqlite3 import and the table creation SQL constant.
2. Import MongitaClientDisk.
3. In the entry point, create a Mongita client instead of a SQLite connection.
4. In add_investment, replace cursor.execute(sql, values) with investments.insert_one(document).
5. In get_investment_value, replace SQL queries with investments.find(filter_doc).

# AVANT (sqlite3)
cursor.execute("INSERT INTO investments VALUES (?, ?, ?, ?, ?);", values)
database.commit()

# APRÈS (Mongita)
investment_document = {
    "coin_id": coin_id,
    "currency": currency,
    "amount": amount,
    "sell": sell,
    "timestamp": datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")
}
investments.insert_one(investment_document)

Note on datetimes: By default, Python cannot encode a datetime in JSON. For Mongita, you must use the string representation (strftime). PyMongo (MongoDB) handles Python datetimes natively.

5.4 Full code — Module 5

05/demos/main.py

import datetime

import requests
import click

from mongita import MongitaClientDisk

def get_coin_price(coin_id, currency):
    url = f"https://api.coingecko.com/api/v3/simple/price?ids={coin_id}&vs_currencies={currency}"
    data = requests.get(url).json()
    coin_price = data[coin_id][currency]
    return coin_price

@click.group()
def cli():
    pass

@click.command()
@click.option("--coin_id", default="bitcoin")
@click.option("--currency", default="usd")
def show_coin_price(coin_id, currency):
    coin_price = get_coin_price(coin_id, currency)
    print(f"The price of {coin_id} is {coin_price:.2f} {currency.upper()}")

@click.command()
@click.option("--coin_id")
@click.option("--currency")
@click.option("--amount", type=float)
@click.option("--sell", is_flag=True)
def add_investment(coin_id, currency, amount, sell):
    investment_document = {
        "coin_id": coin_id,
        "currency": currency,
        "amount": amount,
        "sell": sell,
        "timestamp": datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")
    }
    investments.insert_one(investment_document)

    if sell:
        print(f"Added sell of {amount} {coin_id}")
    else:
        print(f"Added buy of {amount} {coin_id}")

@click.command()
@click.option("--coin_id")
@click.option("--currency")
def get_investment_value(coin_id, currency):
    coin_price = get_coin_price(coin_id, currency)
    buy_result = investments.find({"coin_id": coin_id, "currency": currency, "sell": False})
    sell_result = investments.find({"coin_id": coin_id, "currency": currency, "sell": True})
    buy_amount = sum([doc["amount"] for doc in buy_result])
    sell_amount = sum([doc["amount"] for doc in sell_result])

    total = buy_amount - sell_amount

    print(f"You own a total of {total} {coin_id} worth {total * coin_price} {currency.upper()}")

@click.command()
@click.option("--csv_file")
def import_investments(csv_file):
    pass

cli.add_command(show_coin_price)
cli.add_command(add_investment)
cli.add_command(get_investment_value)
cli.add_command(import_investments)

if __name__ == "__main__":
    client = MongitaClientDisk()
    db = client.portfolio
    investments = db.investments
    cli()

05/demos/requirements.txt

mongita
click 
requests

5.5 Module 5 Summary

• NoSQL refers to any non-relational database. There is no fixed schema, no tables, no foreign keys in the traditional sense.
• NoSQL databases complement relational databases according to the needs of the application.
• Mongita is to MongoDB what SQLite is to PostgreSQL: a local serverless database.
• Mongita documents are Python dictionaries (conceptually JSON objects).
• Mongita implements a subset of the PyMongo API — everything we do here also works with MongoDB.
• CRUD operations use insert_one, find, update_one, delete_one.
• filter documents specify selection criteria; the update documents specify the changes ($inc, $set, etc.).

6. NoSQL database: MongoDB and PyMongo

6.1 MongoDB in Visual Studio Code

MongoDB is the most popular document-oriented database server. The creators of MongoDB maintain an official extension for VS Code that allows you to view and manipulate data directly in the editor.

Installing the extension

1. Open the Extensions panel in VS Code.
2. Search for “MongoDB”.
3. Install “MongoDB for VS Code” published by MongoDB.
4. A sheet icon appears in the sidebar.

Connecting to MongoDB

1. Click on the leaf icon → “Add Connection” button.
2. Click on “Open form” for advanced settings.
3. By default, the connection points to localhost:27017 (standard MongoDB port).
4. Click “Connect”.

MongoDB Playground

The extension includes a MongoDB Playground — a JavaScript editor for interacting with MongoDB:

// Créer une nouvelle base de données et collection
const database = 'ps_database';
const collection = 'ps_collection';
use(database);
db.getCollection(collection).insertOne({...});

6.2 Migration from Mongita to PyMongo

As announced, the migration is trivial — only two lines to modify:

# AVANT (Mongita)
from mongita import MongitaClientDisk

client = MongitaClientDisk()

# APRÈS (PyMongo — MongoDB)
from pymongo import MongoClient

client = MongoClient()  # Connexion locale par défaut (localhost:27017)

The rest of the code is identical. This is proof that Mongita implements a subset of the PyMongo API.

Connection with full connection string (production)

# Connexion avec authentification et paramètres
client = MongoClient("mongodb://username:password@server:27017/database")

Installation

pip install pymongo

6.3 Multi-condition filters

Mongita does not support the $and operator. PyMongo fully supports it.

$and operator

# Trouver les investissements en bitcoin avec un montant > 2
filter_doc = {
    "$and": [
        {"coin_id": "bitcoin"},
        {"amount": {"$gt": 2}}
    ]
}
result = investments.find(filter_doc)

Comparison operators available in PyMongo

Mongita trap: If we use an operator not supported by Mongita (like $and), no error is thrown — the result is simply empty. This violates the Zen of Python: “Errors should not pass silently”. This is one of the reasons to switch to PyMongo for advanced needs.

6.4 Embedded Documents and Lists

MongoDB allows you to store documents in other documents — it’s the equivalent of relationships in a relational ORM, but without foreign keys or joins.

One-to-one relationship with an embedded document

import datetime

metadata = {
    "description": "Coins to buy",
    "currency": "usd",
    "date_created": datetime.datetime.now()  # PyMongo convertit les datetimes !
}

watchlist = {
    "name": "Bulls",
    "metadata": metadata,  # Document imbriqué
    "coins": []
}

watchlists.insert_one(watchlist)

Difference with Mongita: PyMongo automatically converts Python datetime to MongoDB types. With Mongita, you have to use thongs.

The nested document (metadata) does not have its own _id — it belongs to the parent document.

Filtering on an embedded document

We use the dot notation to access the fields of a nested document:

# Trouver les watchlists créées dans les 7 derniers jours
from datetime import datetime, timedelta

one_week_ago = datetime.now() - timedelta(days=7)
filter_doc = {"metadata.date_created": {"$gte": one_week_ago}}
result = watchlists.find(filter_doc)

One-to-many relationship with a list of documents

# Structure avec une liste de sous-documents
watchlist = {
    "name": "Bulls",
    "metadata": {...},
    "coins": [
        {"coin": "bitcoin", "note": "The most popular coin"},
        {"coin": "ethereum", "note": "The second most popular coin"}
    ]
}

$push and $pull operators for lists

# Ajouter un coin à la liste
filter_doc = {"_id": ObjectId(watchlist_id)}
update_doc = {"$push": {"coins": {"coin": "solana", "note": "Hot right now"}}}
watchlists.update_one(filter_doc, update_doc)

# Supprimer un coin de la liste
update_doc = {"$pull": {"coins": {"coin": "solana"}}}
watchlists.update_one(filter_doc, update_doc)

$push and $pull are not supported by Mongita.

6.5 Demo — Embedded Documents and Lists

The demo is organized into several files for clarity:

• utils.py: utility functions (get_coin_prices, seed_data, get_random_datetime).
• manager.py: main CLI application to manage watchlists.

Selecting a document by ObjectId

from bson import ObjectId

# Récupérer tous les noms de watchlists (projection — seulement le champ "name")
watchlist_names = list(watchlists.find({}, {"name": 1}))

# Sélection interactive
for index, wl in enumerate(watchlist_names):
    print(f"{index + 1}: {wl['name']}")
selected_index = int(input("Select a watchlist: ")) - 1
selected_id = watchlist_names[selected_index]["_id"]

# Récupérer le document complet par son ObjectId
watchlist = watchlists.find_one({"_id": ObjectId(selected_id)})

MongoDB automatically assigns an _id of type ObjectId to each document. To reuse it in a filter, you need to wrap it with ObjectId() from the bson library.

6.6 Full code — Module 6

06/demos/mongita_cli.py — Mongita Migration → PyMongo

import datetime

import requests
import click

from pymongo import MongoClient

def get_coin_price(coin_id, currency):
    url = f"https://api.coingecko.com/api/v3/simple/price?ids={coin_id}&vs_currencies={currency}"
    data = requests.get(url).json()
    coin_price = data[coin_id][currency]
    return coin_price

@click.group()
def cli():
    pass

@click.command()
@click.option("--coin_id", default="bitcoin")
@click.option("--currency", default="usd")
def show_coin_price(coin_id, currency):
    coin_price = get_coin_price(coin_id, currency)
    print(f"The price of {coin_id} is {coin_price:.2f} {currency.upper()}")

@click.command()
@click.option("--coin_id")
@click.option("--currency")
@click.option("--amount", type=float)
@click.option("--sell", is_flag=True)
def add_investment(coin_id, currency, amount, sell):
    investment_document = {
        "coin_id": coin_id,
        "currency": currency,
        "amount": amount,
        "sell": sell,
        "timestamp": datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")
    }
    investments.insert_one(investment_document)

    if sell:
        print(f"Added sell of {amount} {coin_id}")
    else:
        print(f"Added buy of {amount} {coin_id}")

@click.command()
@click.option("--coin_id")
@click.option("--currency")
def get_investment_value(coin_id, currency):
    coin_price = get_coin_price(coin_id, currency)
    buy_result = investments.find(
        {"coin_id": coin_id, "currency": currency, "sell": False})
    sell_result = investments.find(
        {"coin_id": coin_id, "currency": currency, "sell": True})
    buy_amount = sum([doc["amount"] for doc in buy_result])
    sell_amount = sum([doc["amount"] for doc in sell_result])

    total = buy_amount - sell_amount

    print(
        f"You own a total of {total} {coin_id} worth {total * coin_price} {currency.upper()}")

@click.command()
@click.option("--csv_file")
def import_investments(csv_file):
    pass

cli.add_command(show_coin_price)
cli.add_command(add_investment)
cli.add_command(get_investment_value)
cli.add_command(import_investments)

if __name__ == "__main__":
    client = MongoClient()
    db = client.portfolio
    investments = db.investments
    cli()

06/demos/utils.py

import datetime
import random

import requests

def get_random_datetime():
    return datetime.datetime.now() - datetime.timedelta(days=random.randint(1, 7))

def get_coin_prices(coin_ids, currency):
    url = f"https://api.coingecko.com/api/v3/simple/price?ids={','.join(coin_ids)}&vs_currencies={currency}"
    data = requests.get(url).json()
    coin_prices = dict([(coin_id, data[coin_id][currency])
                       for coin_id in data])
    return coin_prices

def seed_data(collection):
    collection.insert_many([
        {
            "name": "Bulls",
            "metadata": {
                "description": "Coins to buy",
                "currency": "usd",
                "date_created": get_random_datetime()
            },
            "coins": [
                {"coin": "bitcoin", "note": "The most popular coin",
                 "date_added": get_random_datetime()},
                {"coin": "ethereum", "note": "The second most popular coin",
                 "date_added": get_random_datetime()}
            ]
        },
        {
            "name": "Bears",
            "metadata": {
                "description": "Coins to hold or sell",
                "currency": "usd",
                "date_created": get_random_datetime()
            },
            "coins": [
                {"coin": "solana", "note": "Don't sell this one yet",
                 "date_added": get_random_datetime()}
            ]
        }
    ])

06/demos/manager.py

import datetime

import click
from bson import ObjectId
from pymongo import MongoClient

from utils import get_coin_prices
from utils import seed_data as utils_seed_data

client = MongoClient()

portfolio = client.portfolio

watchlists = portfolio.watchlists

def _get_all_watchlist_names():
    return list(watchlists.find({}, {"name": 1}))

def _select_watchlist(watchlist_names):
    for index, watchlist in enumerate(watchlist_names):
        print(f"{index + 1}: {watchlist['name']}")
    selected_watchlist_index = int(input("Select a watchlist: ")) - 1
    selected_watchlist_id = watchlist_names[selected_watchlist_index]["_id"]
    return watchlists.find_one({"_id": ObjectId(selected_watchlist_id)})

def _select_coin_from_watchlist(watchlist):
    for index, coin in enumerate(watchlist["coins"]):
        print(f"{index + 1}: {coin['coin']}")
    selected_coin_index = int(input("Select a coin: ")) - 1
    return watchlist["coins"][selected_coin_index]

def _add_coin_to_watchlist(watchlist_oid, coin):
    filter = {"_id": ObjectId(watchlist_oid)}
    update = {"$push": {"coins": coin}}
    watchlists.update_one(filter, update)

def _remove_coin_from_watchlist(watchlist_oid, selected_coin):
    filter = {"_id": ObjectId(watchlist_oid)}
    update = {"$pull": {"coins": {"coin": selected_coin}}}
    watchlists.update_one(filter, update)

@click.group()
def cli():
    pass

@click.command(help="Clear the database")
def clear_data():
    portfolio.drop_collection("watchlists")
    print("All data cleared!")

@click.command(help="Seed the database")
@click.option("--force", is_flag=True, help="Seed even if database is not empty")
def seed_data(force):
    if force:
        utils_seed_data(watchlists)
    elif watchlists.count_documents({}) > 0:
        print("The database is not empty, use the --force option or the clear-data command")
    else:
        utils_seed_data(watchlists)

@click.command(help="Add a new watchlist to the portfolio")
@click.option("--name", prompt=True, help="Name of the watchlist")
@click.option("--description", prompt=True, help="Description of the watchlist")
@click.option("--currency", prompt=True, help="Currency to display prices")
def add_watchlist(name, description, currency):
    metadata = {
        "description": description,
        "currency": currency,
        "date_added": datetime.datetime.now()
    }
    watchlist = {
        "name": name,
        "metadata": metadata,
        "coins": []
    }
    watchlists.insert_one(watchlist)
    print(f"Added new {name} watchlist")

@click.command(help="Add a new coin to a watchlist")
@click.option("--coin", prompt=True, help="The coin to add")
@click.option("--note", prompt=True, help="A note")
def add_coin(coin, note):
    selected_watchlist = _select_watchlist(_get_all_watchlist_names())
    _add_coin_to_watchlist(selected_watchlist["_id"], {
        "coin": coin, "note": note, "date_added": datetime.datetime.now()
    })
    print(f"Added {coin} to {selected_watchlist['name']}")

@click.command(help="Remove a coin from a watchlist")
def remove_coin():
    selected_watchlist = _select_watchlist(_get_all_watchlist_names())
    selected_coin = _select_coin_from_watchlist(selected_watchlist)
    _remove_coin_from_watchlist(
        selected_watchlist["_id"], selected_coin["coin"])
    print(f"Removed {selected_coin['coin']} from {selected_watchlist['name']}")

@click.command(help="View the coins and current prices of a watchlist")
def view_watchlist():
    selected_watchlist = _select_watchlist(_get_all_watchlist_names())

    print(
        f"Watchlist: {selected_watchlist['name']} in {selected_watchlist['metadata']['currency']}")
    print(f"{selected_watchlist['metadata']['description']}")
    print(f"{'-' * 25}")
    print("Coins:")
    coin_prices = get_coin_prices(
        [coin["coin"] for coin in selected_watchlist["coins"]],
        selected_watchlist["metadata"]["currency"])
    for index, coin in enumerate(selected_watchlist["coins"]):
        print(
            f"{str(index + 1).rjust(3, ' ')}: {coin['coin']} - {coin['note']}")
        print(f"     Current price: {coin_prices[coin['coin']]}")
    print("Prices provided by CoinGecko")

cli.add_command(add_coin)
cli.add_command(add_watchlist)
cli.add_command(clear_data)
cli.add_command(remove_coin)
cli.add_command(seed_data)
cli.add_command(view_watchlist)

if __name__ == "__main__":
    cli()

06/demos/requirements.txt

pymongo
click 
requests

6.7 Summary of module 6

• PyMongo is the official Python package for accessing MongoDB. Migrating it from Mongita requires only two lines of code.
• The MongoDB for VS Code extension allows you to explore and manipulate MongoDB databases visually.
• The $and operator allows you to aggregate several filter conditions — not supported by Mongita.
• Embedded documents allow you to implement conceptual one-to-one relationships by embedding dictionaries.
• Document lists allow one-to-many relationships to be implemented.
• $push adds a document to a list; $pull removes one — these operators are not available in Mongita.
• PyMongo natively handles Python datetime, unlike Mongita.

7. ODM: MongoEngine

7.1 ODM vs ORM

An ODM (Object Document Mapper) is to MongoDB what an ORM is to relational databases. It creates an association between Python classes and documents in a MongoDB database.

If you’ve used the Django ORM, you’ll immediately feel comfortable with MongoEngine — the APIs are very similar.

MongoEngine doesn’t add anything new to MongoDB — it’s syntactic sugar that simplifies working with the database by eliminating raw Python dictionaries.

7.2 Data modeling with MongoEngine

Base class Document

from mongoengine import Document
from mongoengine import fields

class Investment(Document):
    coin = fields.StringField(max_length=32)
    currency = fields.StringField(max_length=3)
    amount = fields.FloatField(min_value=0.00001)
    timestamp = fields.DateTimeField(default=datetime.datetime.now)
    sell = fields.BooleanField(default=False)

By default, the collection name in MongoDB is the normalized class name in lowercase (here: investment).

Field types available in mongoengine.fields

Best practice: Use from mongoengine import fields and prefix the types with fields. rather than from mongoengine import * (practice not recommended even if present in the official documentation).

Connecting to MongoDB

from mongoengine import connect

# Connexion (crée la base dynamiquement au premier usage)
connect("portfolio_me")  # localhost:27017 par défaut

7.3 Demonstration — Modeling with MongoEngine

Installation

pip install mongoengine click requests

CRUD with MongoEngine

# CREATE — sauvegarder un document
investment = Investment(coin="bitcoin", currency="usd", amount=1.0)
investment.save()

# READ — récupérer tous les documents
all_investments = Investment.objects.all()

# READ — avec filtre (syntaxe Django-like)
bitcoin_investments = Investment.objects(coin="bitcoin")

# READ — un seul document
first = Investment.objects.first()

# UPDATE — modifier et sauvegarder
investment.amount = 2.0
investment.save()

# DELETE — supprimer un document
investment.delete()

# DROP COLLECTION — vider la collection
Investment.drop_collection()

Count documents

count = Investment.objects.count()
if count > 0:
    print("Base non vide")

Method __str__

Important: With MongoEngine, use __str__ rather than __repr__ for display. The __repr__ method does not behave as expected with MongoEngine.

class Investment(Document):
    # ...
    def __str__(self):
        return f"<Investment | coin: {self.coin}, currency: {self.currency}, amount: {self.amount}>"

Partial field selection

# Récupérer uniquement le champ 'coin' (pour les listes de sélection)
investment_coins = Investment.objects.all().fields(coin=1)
for coin in investment_coins:
    print(coin.coin)  # Seul le champ 'coin' est chargé

7.4 Embedded Documents with MongoEngine

EmbeddedDocument vs Document

from mongoengine import Document, EmbeddedDocument
from mongoengine import fields

class WatchlistMetadata(EmbeddedDocument):
    currency = fields.StringField(max_length=3)
    description = fields.StringField()
    date_created = fields.DateField(default=datetime.datetime.now().date)

class WatchlistCoin(EmbeddedDocument):
    coin = fields.StringField(max_length=32)
    note = fields.StringField()
    date_added = fields.DateField(default=datetime.datetime.now().date)

class Watchlist(Document):
    name = fields.StringField(max_length=256)
    metadata = fields.EmbeddedDocumentField(WatchlistMetadata)  # Un-à-un
    coins = fields.EmbeddedDocumentListField(WatchlistCoin)      # Un-à-plusieurs

    def __str__(self):
        return f"<Watchlist name={self.name}, currency={self.metadata.currency} with {len(self.coins)} coin(s)>"

Creating an embedded document

metadata = WatchlistMetadata(currency="usd", description="Coins to watch")
watchlist = Watchlist(name="Bulls", metadata=metadata, coins=[])
watchlist.save()

Adding an item to a list of embedded documents

selected_watchlist.coins.append(WatchlistCoin(coin="bitcoin", note="Number one"))
selected_watchlist.save()  # Sauvegarder le document parent

Queries on embedded documents

MongoEngine uses syntax with double underscore (__) to navigate nested fields (same as Django ORM):

from datetime import datetime, timedelta

# Watchlists créées dans les 7 derniers jours
one_week_ago = datetime.now() - timedelta(days=7)
recent_watchlists = Watchlist.objects(metadata__date_created__gte=one_week_ago.date())

7.5 Demo — Embedded Documents with MongoEngine

The final application watchlists.py (module 7) brings together all the functionalities:

• Investment management.
• Management of watchlists with metadata (WatchlistMetadata) and coin lists (WatchlistCoin).

It implements the commands:

• add-investment: adds an investment.
• view-investment: displays the value of an investment.
• add-watchlist: creates a watchlist with metadata.
• view-watchlist: displays the corners of a watchlist with current prices.
• add-coin: adds a coin to an existing watchlist.
• seed-data / clear-data: initialization and cleanup for development.

7.6 Full code — Module 7

07/demos/manager.py — Managing investments with MongoEngine

import datetime
import random

from mongoengine import connect, Document
from mongoengine import fields
import click

from utils import get_coin_prices

class Investment(Document):
    coin = fields.StringField(max_length=32)
    currency = fields.StringField(max_length=3)
    amount = fields.FloatField(min_value=0.00001)
    timestamp = fields.DateTimeField(default=datetime.datetime.now)
    sell = fields.BooleanField(default=False)

    def __str__(self):
        return f"<Investment | coin: {self.coin}, currency: {self.currency}, amount: {self.amount}>"

def _select_investment():
    investment_coins = Investment.objects.all().fields(coin=1)
    for index, coin in enumerate(investment_coins):
        print(f"{index + 1}: {coin.coin}")
    selected_investment_index = int(input("Select an investment: ")) - 1
    selected_investment_oid = investment_coins[selected_investment_index].id
    return Investment.objects(id=selected_investment_oid).first()

def _seed_data():
    data = [
        ("bitcoin", "USD", 1.0, False),
        ("ethereum", "GBP", 10.0, True),
        ("dogecoin", "EUR", 100.0, False)
    ]

    for row in data:
        Investment(
            coin=row[0],
            currency=row[1],
            amount=row[2],
            sell=row[3],
            timestamp=datetime.datetime.now() - datetime.timedelta(
                days=random.randint(0, 7), minutes=random.randint(0, 60), seconds=random.randint(0, 60)
            )).save()

@click.group()
def cli():
    pass

@click.command(help="Clear the database")
def clear_data():
    Investment.drop_collection()
    print("Cleared data!")

@click.command(help="Seed the database with sample data, use the --force flag to ignore existing data")
@click.option("--force", is_flag=True, default=False)
def seed_data(force):
    if force:
        _seed_data()
    elif Investment.objects.count() > 0:
        print("Data not empty!  Use --force flag to seed database")
    else:
        _seed_data()

@click.command(help="Add a new investment to the portfolio")
@click.option("--coin", prompt=True, help="The name of the coin")
@click.option("--currency", prompt=True, help="The fiat currency to show prices in")
@click.option("--amount", prompt=True, help="The purchase amount")
@click.option("--sell", is_flag=True, default=False, help="If this is a sell (default is False)")
def add_investment(coin, currency, amount, sell):
    investment = Investment(
        coin=coin,
        currency=currency,
        amount=amount,
        sell=sell
    )
    investment.save()

    print(f"Added {'buy' if not sell else 'sell'} for {amount} {coin} in {currency}")

@click.command(help="See the details of an investment")
def view_investment():
    selected_investment = _select_investment()
    coin_price = get_coin_prices([selected_investment.coin], selected_investment.currency.lower())[
        selected_investment.coin]
    print(f"You {'bought' if not selected_investment.sell else 'sold'} {selected_investment.amount} {selected_investment.coin} for {coin_price * selected_investment.amount} {selected_investment.currency}")

cli.add_command(add_investment)
cli.add_command(clear_data)
cli.add_command(seed_data)
cli.add_command(view_investment)

if __name__ == "__main__":
    connect("portfolio_me")
    cli()

07/demos/watchlists.py — Complete application with embedded documents

import datetime
import random

from mongoengine import connect, Document, EmbeddedDocument
from mongoengine import fields
import click

from utils import get_coin_prices

class Investment(Document):
    coin = fields.StringField(max_length=32)
    currency = fields.StringField(max_length=3)
    amount = fields.FloatField(min_value=0.00001)
    timestamp = fields.DateTimeField(default=datetime.datetime.now)
    sell = fields.BooleanField(default=False)

    def __str__(self):
        return f"<Investment | coin: {self.coin}, currency: {self.currency}, amount: {self.amount}>"

class WatchlistMetadata(EmbeddedDocument):
    currency = fields.StringField(max_length=3)
    description = fields.StringField()
    date_created = fields.DateField(default=datetime.datetime.now().date)

class WatchlistCoin(EmbeddedDocument):
    coin = fields.StringField(max_length=32)
    note = fields.StringField()
    date_added = fields.DateField(default=datetime.datetime.now().date)

class Watchlist(Document):
    name = fields.StringField(max_length=256)
    metadata = fields.EmbeddedDocumentField(WatchlistMetadata)
    coins = fields.EmbeddedDocumentListField(WatchlistCoin)

    def __str__(self):
        return f"<Watchlist name={self.name}, currency={self.metadata.currency} with {len(self.coins)} coin(s)>"

def _select_investment():
    investment_coins = Investment.objects.all().fields(coin=1)
    for index, coin in enumerate(investment_coins):
        print(f"{index + 1}: {coin.coin}")
    selected_investment_index = int(input("Select an investment: ")) - 1
    selected_investment_oid = investment_coins[selected_investment_index].id
    return Investment.objects(id=selected_investment_oid).first()

def _select_watchlist():
    watchlist_names = Watchlist.objects.all().fields(name=1)
    for index, name in enumerate(watchlist_names):
        print(f"{index + 1}: {name.name}")
    selected_watchlist_index = int(input("Select a watchlist: ")) - 1
    selected_watchlist_oid = watchlist_names[selected_watchlist_index].id
    return Watchlist.objects(id=selected_watchlist_oid).first()

def _seed_data():
    data = [
        ("bitcoin", "USD", 1.0, False),
        ("ethereum", "GBP", 10.0, True),
        ("dogecoin", "EUR", 100.0, False)
    ]

    watchlist_data = [
        ("Bulls", "Coins to buy", "USD", [
         ("bitcoin", "Bitcoin is number one!"), ("ethereum", "Ethereum is number two!")]),
        ("Bears", "Coins to sell", "GBP", [("solana", "Meh ...")])
    ]

    for row in data:
        Investment(
            coin=row[0],
            currency=row[1],
            amount=row[2],
            sell=row[3],
            timestamp=datetime.datetime.now() - datetime.timedelta(
                days=random.randint(0, 7), minutes=random.randint(0, 60), seconds=random.randint(0, 60)
            )).save()

    for row in watchlist_data:
        Watchlist(
            name=row[0],
            metadata=WatchlistMetadata(description=row[1], currency=row[2]),
            coins=[WatchlistCoin(coin=coin[0], note=coin[1])
                   for coin in row[3]]
        ).save()

@click.group()
def cli():
    pass

@click.command(help="Clear the database")
def clear_data():
    Investment.drop_collection()
    Watchlist.drop_collection()
    print("Cleared data!")

@click.command(help="Seed the database with sample data, use the --force flag to ignore existing data")
@click.option("--force", is_flag=True, default=False)
def seed_data(force):
    if force:
        _seed_data()
    elif Investment.objects.count() > 0:
        print("Data not empty!  Use --force flag to seed database")
    else:
        _seed_data()

@click.command(help="Add a new investment to the portfolio")
@click.option("--coin", prompt=True, help="The name of the coin")
@click.option("--currency", prompt=True, help="The fiat currency to show prices in")
@click.option("--amount", prompt=True, help="The purchase amount")
@click.option("--sell", is_flag=True, default=False, help="If this is a sell (default is False)")
def add_investment(coin, currency, amount, sell):
    investment = Investment(
        coin=coin,
        currency=currency,
        amount=amount,
        sell=sell
    )
    investment.save()

    print(f"Added {'buy' if not sell else 'sell'} for {amount} {coin} in {currency}")

@click.command(help="See the details of an investment")
def view_investment():
    selected_investment = _select_investment()
    coin_price = get_coin_prices([selected_investment.coin], selected_investment.currency.lower())[
        selected_investment.coin]
    print(f"You {'bought' if not selected_investment.sell else 'sold'} {selected_investment.amount} {selected_investment.coin} for {coin_price * selected_investment.amount} {selected_investment.currency}")

@click.command(help="Add a new watchlist to the portfolio")
@click.option("--name", help="The name of the watchlist", prompt=True)
@click.option("--description", help="The description of the watchlist", prompt=True)
@click.option("--currency", help="The currency to display coin prices in", prompt=True)
def add_watchlist(name, description, currency):
    metadata = WatchlistMetadata(currency=currency, description=description)
    watchlist = Watchlist(name=name, metadata=metadata, coins=[])
    watchlist.save()
    print(f"Added watchlist {name}")

@click.command(help="View the coins in a watchlist")
def view_watchlist():
    selected_watchlist = _select_watchlist()
    coins = [coin.coin for coin in selected_watchlist.coins]
    coin_prices = get_coin_prices(
        coins, selected_watchlist.metadata.currency.lower())
    print(
        f"Watchlist: {selected_watchlist.name} in {selected_watchlist.metadata.currency}")
    print(f"{selected_watchlist.metadata.description}")
    print("Coins: ")
    for index, coin in enumerate(coins):
        print(f"{index + 1}: {coin} | {coin_prices[coin]}")
    print("Prices provided by CoinGecko")

@click.command(help="Add a coin to a watchlist")
@click.option("--coin", help="The coin to add", prompt=True)
@click.option("--note", help="A note", prompt=True)
def add_coin(coin, note):
    selected_watchlist = _select_watchlist()
    selected_watchlist.coins.append(
        WatchlistCoin(coin=coin, note=note)
    )
    selected_watchlist.save()
    print(f"Added {coin} to {selected_watchlist.name}")

cli.add_command(add_coin)
cli.add_command(add_investment)
cli.add_command(add_watchlist)
cli.add_command(clear_data)
cli.add_command(seed_data)
cli.add_command(view_watchlist)
cli.add_command(view_investment)

if __name__ == "__main__":
    connect("portfolio_me")
    cli()

07/demos/utils.py

import requests

def get_coin_prices(coin_ids, currency):
    url = f"https://api.coingecko.com/api/v3/simple/price?ids={','.join(coin_ids)}&vs_currencies={currency}"
    data = requests.get(url).json()
    coin_prices = dict([(coin_id, data[coin_id][currency])
                       for coin_id in data])
    return coin_prices

07/demos/requirements.txt

mongoengine
requests
click

7.7 Module 7 Summary

• MongoEngine is an ODM (Object Document Mapper) for MongoDB.
• It creates a mapping between Python classes and MongoDB documents.
• Its API is inspired by the Django ORM — familiar to Django developers.
• Model classes inherit from Document; Embedded documents inherit from EmbeddedDocument.
• Fields are defined via mongoengine.fields types.
• EmbeddedDocumentField models one-to-one relationships; EmbeddedDocumentListField models one-to-many relationships.
• Queries filtering on embedded documents use the __ (double underscore) syntax.
• MongoEngine is syntactic sugar: it does not modify the data stored in MongoDB.
• Use __str__ rather than __repr__ with MongoEngine document classes.

8. General summary and choosing the right solution

The course covered six technologies organized along two axes: database type (relational vs NoSQL) and abstraction level (local/file → low level → high level).

Overall comparison table

Decision tree

Quel type de données ?
├── Données structurées, schéma fixe, relations → Relationnelle
│   ├── Local, sans serveur, prototypage → SQLite + sqlite3
│   └── Serveur, production
│       ├── SQL direct, contrôle fin → PostgreSQL + psycopg2
│       └── ORM, code Python pur, multi-DB → SQLAlchemy
└── Données flexibles, JSON, schéma variable → NoSQL document
    ├── Local, sans serveur, tests → Mongita
    └── Serveur MongoDB
        ├── Accès direct, dictionnaires Python → PyMongo
        └── ODM, classes Python, API Django-like → MongoEngine

Key Takeaways

1. Relational bases vs NoSQL: they complement each other — the choice depends on the needs of the application, not a fashion.

2. SQLite for development, PostgreSQL for production: with SQLAlchemy, switching from one to the other only requires changing the connection string.

3. Parameterized queries always: never construct SQL queries with f-strings from user data — use placeholders (? for SQLite, %s for psycopg2).

4. Row factories and dataclasses: prefer typed Python objects to raw tuples for maintainability.

5. Mongita ⊆ PyMongo: everything you learn with Mongita can be reused with PyMongo — with two fewer lines of code.

6. MongoEngine ≈ Django ORM: if you know the Django ORM, MongoEngine is almost immediately masterable.

7. Close connections: with psycopg2, always close the cursor and the connection after use. With SQLAlchemy and its Session, use the context manager with Session(engine) as session:.

Installation commands per module

# Module 2 — SQLite (inclus dans Python)
pip install requests click

# Module 3 — PostgreSQL
pip install psycopg2-binary click requests

# Module 4 — SQLAlchemy ORM
pip install sqlalchemy psycopg2-binary click requests

# Module 5 — Mongita
pip install mongita click requests

# Module 6 — PyMongo
pip install pymongo click requests

# Module 7 — MongoEngine
pip install mongoengine click requests

Using command-line applications

# Module 2 — SQLite
python main.py show-coin-price --coin_id=bitcoin --currency=usd
python main.py add-investment --coin_id=bitcoin --currency=usd --amount=1.0
python main.py get-investment-value --coin_id=bitcoin --currency=usd
python main.py import-investments --csv_file=import.csv

# Module 3 — PostgreSQL
python manager.py new-investment
python manager.py import-investments --filename=import.csv
python manager.py view-investments --currency=usd

# Module 4 — SQLAlchemy
python manager.py add-portfolio
python manager.py add-investment
python manager.py view-portfolio
python manager.py clear-database

# Module 5 — Mongita
python main.py show-coin-price
python main.py add-investment --coin_id=bitcoin --currency=usd --amount=1.0
python main.py get-investment-value --coin_id=bitcoin --currency=usd

# Module 6 — PyMongo (watchlists)
python manager.py seed-data
python manager.py add-watchlist
python manager.py add-coin
python manager.py view-watchlist
python manager.py remove-coin
python manager.py clear-data

# Module 7 — MongoEngine
python watchlists.py seed-data
python watchlists.py add-investment
python watchlists.py add-watchlist
python watchlists.py add-coin
python watchlists.py view-watchlist
python watchlists.py view-investment
python watchlists.py clear-data

Search Terms

databases · python · foundations · data · analysis · engineering · analytics · documents · embedded · mongoengine · application · database · demonstration · document · nosql · requirements.txt · mongita · mongodb · row · sqlite · factories · insertion · installation · manager.py