Peewee

---

1. Modeling a simple blog post:

from peewee import *

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()

    class Meta:
        database = db

1. Adding a foreign key relationship:

from peewee import *

class User(Model):
    name = CharField()

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()
    user = ForeignKeyField(User, backref='posts')

    class Meta:
        database = db

1. Creating a many-to-many relationship:

from peewee import *

class Tag(Model):
    name = CharField()

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()
    tags = ManyToManyField(Tag, backref='posts')

    class Meta:
        database = db

1. Adding a composite key:

from peewee import *

class Tag(Model):
    name = CharField()
    slug = CharField()

    def __composite_key__(self):
        return (self.name, self.slug)

    class Meta:
        database = db

1. Creating a self-referencing relationship:

from peewee import *

class Category(Model):
    name = CharField()
    parent = ForeignKeyField('self', null=True, backref='children')

    class Meta:
        database = db

1. Adding a database constraint:

from peewee import *

class User(Model):
    name = CharField(unique=True)

    class Meta:
        database = db

1. Using a custom field type:

import datetime

from peewee import *

class MyField(Field):
    def __init__(self):
        super(MyField, self).__init__()

    def coerce(self, value):
        if isinstance(value, str):
            return datetime.datetime.strptime(value, '%Y-%m-%d')
        return value

class MyModel(Model):
    date = MyField()

    class Meta:
        database = db

1. Using a custom function:

from peewee import *

import math

def distance(lat1, lon1, lat2, lon2):
    """Calculates the distance between two points in kilometers."""
    return math.sqrt((lat1 - lat2) ** 2 + (lon1 - lon2) ** 2) * 40000 / 360

class Location(Model):
    latitude = FloatField()
    longitude = FloatField()

    def distance_to(self, other_location):
        return distance(self.latitude, self.longitude, other_location.latitude, other_location.longitude)

    class Meta:
        database = db

1. Creating an aggregate query:

from peewee import *

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()

    class Meta:
        database = db

# Count the number of posts in the database
Post.select(fn.COUNT('*')).scalar()

1. Using a subquery:

from peewee import *

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()

    class Meta:
        database = db

# Find all posts that were created before the most recent post
Post.select().where(Post.timestamp < (Post.select(fn.MAX(Post.timestamp)).scalar()))

1. Creating a view:

from peewee import *

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()

    class Meta:
        database = db

# Create a view that shows the title and timestamp of all posts
db.execute_sql('CREATE VIEW post_summary AS SELECT title, timestamp FROM post')

# Query the view
PostSummary = peewee.Model.from_query('post_summary')
PostSummary.select().execute()

1. Using a trigger:

from peewee import *

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()

    class Meta:
        database = db

# Create a trigger that updates the timestamp when a post is updated
db.execute_sql('CREATE TRIGGER update_timestamp BEFORE UPDATE ON post FOR EACH ROW SET new.timestamp = DATETIME("now")')

1. Creating a stored procedure:

from peewee import *

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()

    class Meta:
        database = db

# Create a stored procedure that returns the number of posts in the database
db.execute_sql('CREATE FUNCTION get_post_count() RETURNS INTEGER AS $$BEGIN RETURN (SELECT COUNT(*) FROM post); END$$;')

# Call the stored procedure
db.execute_sql('SELECT get_post_count()')

1. Using a database adapter:

import peewee
from peewee import MySQLDatabase

database = MySQLDatabase('my_database', user='root', password='password')

class Post(peewee.Model):
    title = peewee.CharField()
    body = peewee.TextField()
    timestamp = peewee.DateTimeField()

    class Meta:
        database = database

1. Using a connection pool:

from peewee import *

# Create a connection pool
pool = PooledMySQLDatabase('my_database', user='root', password='password', max_connections=10)

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()

    class Meta:
        database = pool

1. Using a model manager:

from peewee import *

class UserManager(ModelManager):
    def active_users(self):
        return self.filter(is_active=True)

class User(Model):
    is_active = BooleanField()

    class Meta:
        manager_class = UserManager

1. Using a custom validator:

from peewee import *

class EmailValidator(Validator):
    def __init__(self):
        super(EmailValidator, self).__init__()

    def validate(self, value):
        if not re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', value):
            raise ValueError('Invalid email address')

class User(Model):
    email = CharField(validators=[EmailValidator()])

    class Meta:
        database = db

1. Using a custom hook:

from peewee import *

class LoggingHook(Hook):
    def before_query(self, model, query):
        print('Query: {}'.format(query))

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()

    class Meta:
        hooks = [LoggingHook()]

1. Using a custom loader:

from peewee import *

class JSONLoader(Loader):
    def load_from_file(self, filename):
        with open(filename, 'r') as f:
            return json.load(f)

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()

    class Meta:
        loader_class = JSONLoader

1. Using a custom writer:

from peewee import *

class JSONWriter(Writer):
    def save_to_file(self, model, filename):
        with open(filename, 'w') as f:
            json.dump(model, f)

class Post(Model):
    title = CharField()
    body = TextField()
    timestamp = DateTimeField()

    class Meta:
        writer_class = JSONWriter

1. Interacting with a database in a multithreaded environment:

import threading

from peewee import *

# Create a database object that can be shared across threads
database = SqliteDatabase('my_database.db')

def thread_function():
    with database.atomic() as txn:
        # Perform database operations within this block
        pass

# Create and start a thread
thread = threading.Thread(target=thread_function)
thread.start()

1. Using a SQLite BLOB field:

from peewee import *

class Document(Model):
    name = CharField()
    content = BlobField()

    class Meta:
        database = db

1. Using a PostgreSQL JSON field:

from peewee import *

class Settings(Model):
    key = CharField(primary_key=True)
    value = JSONField()

    class Meta:
        database = db

1. Using a MySQL TEXT field:

from peewee import *

class Article(Model):
    title = CharField()
    body = TextField()

    class Meta:
        database = db

1. Using a Microsoft SQL Server VARCHAR field:

from peewee import *

class Product(Model):
    name = CharField(max_length=50)
    description = TextField()

    class Meta:
        database = db

1. Adding a column to a model:

from peewee import *

class User(Model):
    name = CharField()
    email = CharField()

    class Meta:
        database = db

User.add_field('age', IntegerField())

1. Dropping a column from a model:

from peewee import *

class User(Model):
    name = CharField()
    email = CharField()
    age = IntegerField()

    class Meta:
        database = db

User.drop_field('age')

1. Renaming a column in a model:

from peewee import *

class User(Model):
    name = CharField()
    email = CharField()

    class Meta:
        database = db

User.rename_field('email', 'username')

1. Creating a custom Peewee field type:

from peewee import *

class MyField(Field):
    def __init__(self, *args, **kwargs):
        super(MyField, self).__init__(*args, **kwargs)

        self.validators.append(MyValidator())

    def db_value(self, value):
        return value + 1

class MyModel(Model):
    value = MyField()

    class Meta:
        database = db

1. Using a custom Peewee validator:

from peewee import *

class MyValidator(Validator):
    def validate(self, value):
        if value < 0:
            raise ValueError('Value must be non-negative')

class MyModel(Model):
    value = IntegerField(validators=[MyValidator()])

    class Meta:
        database = db

1. Creating a custom Peewee hook:

from peewee import *

class MyHook(Hook):
    def before_query(self, model, query):
        print('Query: {}'.format(query))

class MyModel(Model):
    value = CharField()

    class Meta:
        database = db
        hooks = [MyHook()]

1. Creating a custom Peewee loader:

from peewee import *

import json

class MyLoader(Loader):
    def load_from_file(self, filename):
        with open(filename, 'r') as f:
            return json.load(f)

class MyModel(Model):
    value = CharField()

    class Meta:
        database = db
        loader_class = MyLoader

1. Creating a custom Peewee writer:

from peewee import *

import json

class MyWriter(Writer):
    def save_to_file(self, model, filename):
        with open(filename, 'w') as f:
            json.dump(model, f)

class MyModel(Model):
    value = CharField()

    class Meta:
        database = db
        writer_class = MyWriter

1. Using a Peewee model manager:

from peewee import *

class MyManager(ModelManager):
    def get_by_name(self, name):
        return self.get(name=name)

class MyModel(Model):
    name = CharField()

    class Meta:
        database = db
        manager_class = MyManager

1. Using a Peewee queryset

from peewee import *

class User(Model):
    name = CharField()

    class Meta:
        database = db

for user in User.select().where(User.name.contains('a')):
    print(user.name)

1. Using a Peewee subquery:

from peewee import *

class Post(Model):
    title = CharField()
    author = ForeignKeyField(User, backref='posts')

    class Meta:
        database = db

query = (Post
        .select(Post.title, User.name)
        .join(User)
        .where(User.id == 1)
)

for post in query:
    print(post.title, post.author)

1. Using a Peewee aggregation

from peewee import *

class Post(Model):
    title = CharField()
    author = ForeignKeyField(User, backref='posts')

    class Meta:
        database = db

query = (Post
        .select(fn.COUNT(Post.id).alias('num_posts'), User.name)
        .join(User)
        .group_by(User.name)
)

for row in query:
    print(row.num_posts, row.name)

1. Using a Peewee raw query

from peewee import *

class User(Model):
    name = CharField()

    class Meta:
        database = db

query = db.execute_sql('SELECT * FROM user WHERE name LIKE ?', ('%a%',))

for user in query:
    print(user.name)

1. Using a Peewee transaction

from peewee import *

class User(Model):
    name = CharField()

    class Meta:
        database = db

with db.atomic() as txn:
    user1 = User.create(name='Alice')
    user2 = User.create(name='Bob')

1. Using a Peewee model hook

from peewee import *

class User(Model):
    name = CharField()

    class Meta:
        database = db

    def before_save(self, sender, instance):
        print('User {} is being saved.'.format(instance.name))

1. Using a Peewee validator

from peewee import *

class User(Model):
    name = CharField(validators=[validators.MinLengthValidator(3)])

    class Meta:
        database = db

1. Using a Peewee loader

from peewee import *

class User(Model):
    name = CharField()

    class Meta:
        database = db
        loader_class = JSONLoader

1. Using a Peewee writer

from peewee import *

class User(Model):
    name = CharField()

    class Meta:
        database = db
        writer_class = JSONWriter

1. Using a Peewee manager

from peewee import *

class UserManager(ModelManager):
    def get_by_name(self, name):
        return self.get(name=name)

class User(Model):
    name = CharField()

    class Meta:
        database = db
        manager_class = UserManager

1. Using a Peewee foreign key

from peewee import *

class Post(Model):
    title = CharField()
    author = ForeignKeyField(User, backref='posts')

    class Meta:
        database = db

1. Using a Peewee one-to-many relationship

from peewee import *

class User(Model):
    name = CharField()

class Post(Model):
    title = CharField()
    author = ForeignKeyField(User, backref='posts')

    class Meta:
        database = db

1. Using a Peewee many-to-many relationship

from peewee import *

class User(Model):
    name = CharField()

class Tag(Model):
    name = CharField()

class Post(Model):
    title = CharField()
    author = ForeignKeyField(User, backref='posts')
    tags = ManyToManyField(Tag, backref='posts')

    class Meta:
        database = db

1. Using a Peewee composite key

from peewee import *

class User(Model):
    first_name = CharField()
    last_name = CharField()

    class Meta:
        database = db
        primary_key = CompositeKey('first_name', 'last_name')

1. Using a Peewee database constraint

from peewee import *

class User(Model):
    name = CharField(unique=True)

    class Meta:
        database = db

1. Using a Peewee custom field type

from peewee import *

class ColorField(Field):
    def db_value(self, value):
        return value.name

    def python_value(self, value):
        return Color(value)

class User(Model):
    name = CharField()
    favorite_color = ColorField()

    class Meta:
        database = db

1. Using a Peewee custom function

from peewee import *

def distance(lat1, lon1, lat2, lon2):
    """Calculates the distance between two points in kilometers."""
    return math.sqrt((lat1 - lat2) ** 2 + (lon1 - lon2) ** 2) * 40000 / 360

class Location(Model):
    latitude = FloatField()
    longitude = FloatField()

    def distance_to(self, other_location):
        return distance(self.latitude, self.longitude, other_location.latitude, other_location.longitude)

    class Meta:
        database = db

1. Using a Peewee aggregate query

from peewee import *

class Post(Model):
    title = CharField()
    author = ForeignKeyField(User, backref='posts')

    class Meta:
        database = db

# Count the number of posts in the database
Post.select(fn.COUNT('*')).scalar()

1. Using a Peewee subquery

from peewee import *

class Post(Model):
    title = CharField()
    author = ForeignKeyField(User, backref='posts')

    class Meta:
        database = db

# Find all posts that were created before the most recent post
Post.select().where(Post.timestamp < (Post.select(fn.MAX(Post.timestamp)).scalar()))

1. Using a Peewee view

from peewee import *

class Post(Model):
    title = CharField()
    author = ForeignKeyField(User, backref='posts')

    class Meta:
        database = db

# Create a view that shows the title and timestamp of all posts
db.execute_sql('CREATE VIEW post_summary AS SELECT title, timestamp FROM post')

# Query the view
PostSummary = peewee.Model.from_query('post_summary')
PostSummary.select().execute()

5