tessl/pypi-peewee

A lightweight Python ORM (Object-Relational Mapping) library that provides a simple and expressive interface for database operations.

—

Pending

Overview

Eval results

Files

Queries and Operations

Name: tessl/pypi-peewee
Author: tessl

Query building, filtering, joins, aggregation, and bulk operations. Peewee provides a fluent interface for constructing complex SQL queries using Python syntax with type safety and database portability.

Capabilities

Select Queries

SELECT query building with filtering, joins, grouping, ordering, and result iteration. The core interface for retrieving data from the database.

class Select:
    """
    SELECT query builder.
    """
    def __init__(self, from_list=None, columns=None, group_by=None,
                 having=None, distinct=None, windows=None, for_update=None,
                 for_update_of=None, nowait=None, lateral=None, **kwargs):
        """
        Initialize SELECT query.
        
        Parameters:
        - from_list: Tables to select from
        - columns: Columns to select
        - group_by: GROUP BY expressions
        - having: HAVING conditions
        - distinct: DISTINCT clause
        - windows: Window functions
        - for_update: FOR UPDATE locking
        - for_update_of: FOR UPDATE OF specific tables
        - nowait: NOWAIT option for locking
        - lateral: LATERAL join support
        - **kwargs: Additional query options
        """
    
    def where(self, *expressions):
        """
        Add WHERE conditions to the query.
        
        Parameters:
        - *expressions: Query expressions to filter by
        
        Returns:
        Select: Query with added conditions
        """
    
    def join(self, dest, join_type='INNER', on=None):
        """
        Add JOIN clause to the query.
        
        Parameters:
        - dest: Model or table to join
        - join_type (str): JOIN type ('INNER', 'LEFT', 'RIGHT', 'FULL')
        - on: Join condition (auto-detected if None)
        
        Returns:
        Select: Query with added join
        """
    
    def switch(self, dest=None):
        """
        Switch query context for chained joins.
        
        Parameters:
        - dest: Model to switch to (None for original)
        
        Returns:
        Select: Query with switched context
        """
    
    def group_by(self, *columns):
        """
        Add GROUP BY clause to the query.
        
        Parameters:
        - *columns: Columns to group by
        
        Returns:
        Select: Query with grouping
        """
    
    def having(self, *expressions):
        """
        Add HAVING conditions to grouped query.
        
        Parameters:
        - *expressions: Having conditions
        
        Returns:
        Select: Query with having conditions
        """
    
    def order_by(self, *columns):
        """
        Add ORDER BY clause to the query.
        
        Parameters:
        - *columns: Columns to order by (use .desc() for descending)
        
        Returns:
        Select: Query with ordering
        """
    
    def limit(self, limit, offset=0):
        """
        Add LIMIT and OFFSET to the query.
        
        Parameters:
        - limit (int): Maximum number of results
        - offset (int): Number of results to skip
        
        Returns:
        Select: Query with limit/offset
        """
    
    def offset(self, offset):
        """
        Add OFFSET to the query.
        
        Parameters:
        - offset (int): Number of results to skip
        
        Returns:
        Select: Query with offset
        """
    
    def paginate(self, page, paginate_by=20):
        """
        Paginate query results.
        
        Parameters:
        - page (int): Page number (1-based)
        - paginate_by (int): Results per page
        
        Returns:
        Select: Query with pagination
        """
    
    def distinct(self, distinct=True):
        """
        Make query return distinct results.
        
        Parameters:
        - distinct (bool): Enable/disable distinct
        
        Returns:
        Select: Query with distinct
        """
    
    def aggregate(self, aggregation):
        """
        Perform aggregation on query results.
        
        Parameters:
        - aggregation: Aggregation expression (e.g., fn.COUNT())
        
        Returns:
        Scalar result of aggregation
        """
    
    def count(self, clear_limit=False):
        """
        Count query results.
        
        Parameters:
        - clear_limit (bool): Remove LIMIT for accurate count
        
        Returns:
        int: Number of matching results
        """
    
    def exists(self):
        """
        Check if query has any results.
        
        Returns:
        bool: True if results exist
        """
    
    def get(self):
        """
        Get single result from query.
        
        Returns:
        Model instance
        
        Raises:
        DoesNotExist: If no results found
        MultipleObjectsReturned: If multiple results found
        """
    
    def first(self, n=1):
        """
        Get first n results.
        
        Parameters:
        - n (int): Number of results to return
        
        Returns:
        Model instance (n=1) or list of instances (n>1)
        """
    
    def scalar(self, as_tuple=False):
        """
        Get scalar value from query.
        
        Parameters:
        - as_tuple (bool): Return as tuple if multiple columns
        
        Returns:
        Scalar value or tuple
        """

Usage examples:

from peewee import *

# Basic select
users = User.select().where(User.age > 18)

# Select specific fields
users = User.select(User.username, User.email).where(User.is_active == True)

# Joins
query = (User
         .select(User.username, fn.COUNT(Post.id).alias('post_count'))
         .join(Post, JOIN.LEFT_OUTER)
         .group_by(User.username)
         .order_by(fn.COUNT(Post.id).desc()))

# Complex filtering
users = (User
         .select()
         .where(
             (User.age.between(18, 65)) &
             (User.email.contains('@gmail.com')) |
             (User.username.in_(['admin', 'moderator']))
         )
         .order_by(User.created_at.desc())
         .limit(10))

# Pagination
page_1 = User.select().paginate(1, 20)  # First 20 users
page_2 = User.select().paginate(2, 20)  # Next 20 users

# Aggregation
total_users = User.select().count()
avg_age = User.select(fn.AVG(User.age)).scalar()

Query Expressions and Operators

Query building components for constructing WHERE conditions, field operations, and SQL expressions.

class Field:
    """
    Field query operations.
    """
    def __eq__(self, other):
        """Equality comparison (=)."""
    
    def __ne__(self, other):
        """Not equal comparison (!=)."""
    
    def __lt__(self, other):
        """Less than comparison (<)."""
    
    def __le__(self, other):
        """Less than or equal (<=)."""
    
    def __gt__(self, other):
        """Greater than comparison (>)."""
    
    def __ge__(self, other):
        """Greater than or equal (>=)."""
    
    def __lshift__(self, other):
        """Left shift for special operations."""
    
    def __rshift__(self, other):
        """Right shift for special operations."""
    
    def in_(self, values):
        """IN clause with list of values."""
    
    def not_in(self, values):
        """NOT IN clause with list of values."""
    
    def is_null(self, null=True):
        """IS NULL / IS NOT NULL check."""
    
    def contains(self, value):
        """String contains (LIKE %value%)."""
    
    def startswith(self, value):
        """String starts with (LIKE value%)."""
    
    def endswith(self, value):
        """String ends with (LIKE %value)."""
    
    def between(self, low, high):
        """BETWEEN low AND high clause."""
    
    def regexp(self, pattern):
        """Regular expression match."""
    
    def concat(self, *args):
        """String concatenation."""
    
    def desc(self):
        """Descending order modifier."""
    
    def asc(self):
        """Ascending order modifier."""

# Logical operators
def __and__(self, other):
    """AND logical operator (&)."""

def __or__(self, other):
    """OR logical operator (|)."""

def __invert__(self):
    """NOT logical operator (~)."""

Usage examples:

# Comparison operators
User.select().where(User.age > 18)
User.select().where(User.age >= 21)
User.select().where(User.username == 'john')
User.select().where(User.email != 'test@example.com')

# IN and NOT IN
User.select().where(User.username.in_(['john', 'jane', 'bob']))
User.select().where(User.status.not_in(['banned', 'suspended']))

# NULL checks
User.select().where(User.email.is_null(False))  # IS NOT NULL
User.select().where(User.deleted_at.is_null())  # IS NULL

# String operations
User.select().where(User.email.contains('@gmail.com'))
User.select().where(User.username.startswith('admin'))
User.select().where(User.phone.endswith('1234'))

# Range operations
User.select().where(User.age.between(18, 65))
Post.select().where(Post.created_at.between(start_date, end_date))

# Logical combinations
query = User.select().where(
    (User.age > 18) & 
    (User.is_active == True) |
    (User.role == 'admin')
)

# Complex expressions
query = User.select().where(
    (User.age.between(18, 65)) &
    ((User.email.contains('@gmail.com')) | (User.email.contains('@yahoo.com'))) &
    ~(User.username.in_(['spam', 'test']))
)

Function Calls and Expressions

SQL function calls, mathematical operations, and custom expressions for advanced query capabilities.

class fn:
    """
    SQL function call builder.
    """
    # Aggregate functions
    COUNT = lambda *args: ...      # COUNT(*)
    SUM = lambda field: ...        # SUM(field)
    AVG = lambda field: ...        # AVG(field)
    MIN = lambda field: ...        # MIN(field)
    MAX = lambda field: ...        # MAX(field)
    
    # String functions
    LOWER = lambda field: ...      # LOWER(field)
    UPPER = lambda field: ...      # UPPER(field)
    LENGTH = lambda field: ...     # LENGTH(field)
    SUBSTR = lambda field, start, length=None: ...  # SUBSTR(field, start, length)
    
    # Date functions
    NOW = lambda: ...              # NOW()
    DATE = lambda field: ...       # DATE(field)
    YEAR = lambda field: ...       # YEAR(field)
    MONTH = lambda field: ...      # MONTH(field)
    DAY = lambda field: ...        # DAY(field)
    
    # Math functions
    ABS = lambda field: ...        # ABS(field)
    ROUND = lambda field, precision=None: ...  # ROUND(field, precision)
    
    # Conditional functions
    COALESCE = lambda *args: ...   # COALESCE(arg1, arg2, ...)
    
    # Custom functions
    def __getattr__(self, name):
        """Call any SQL function by name."""

class Case:
    """
    SQL CASE expression builder.
    """
    def __init__(self, predicate=None, expression_tuples=None, default=None):
        """
        Parameters:
        - predicate: Field to switch on (for simple CASE)
        - expression_tuples: List of (condition, result) tuples
        - default: Default value for ELSE clause
        """
    
    def when(self, expression, value):
        """
        Add WHEN clause.
        
        Parameters:
        - expression: Condition to test
        - value: Value to return if condition is true
        
        Returns:
        Case: Updated case expression
        """
    
    def else_(self, value):
        """
        Add ELSE clause.
        
        Parameters:
        - value: Default value
        
        Returns:
        Case: Case expression with default
        """

class Cast:
    """
    SQL CAST expression for type conversion.
    """
    def __init__(self, field, cast):
        """
        Parameters:
        - field: Field to cast
        - cast (str): Target data type
        """

class SQL:
    """
    Raw SQL expression wrapper.
    """
    def __init__(self, sql, *params):
        """
        Parameters:
        - sql (str): Raw SQL string
        - *params: Parameter values for placeholders
        """

Usage examples:

from peewee import *

# Aggregate functions
user_count = User.select(fn.COUNT(User.id)).scalar()
total_age = User.select(fn.SUM(User.age)).scalar()
avg_age = User.select(fn.AVG(User.age)).scalar()

# String functions
users = User.select().where(fn.LOWER(User.username) == 'john')
users_with_lengths = User.select(User.username, fn.LENGTH(User.username))

# Date functions
recent_posts = Post.select().where(Post.created_at >= fn.NOW() - timedelta(days=7))
posts_by_year = (Post
                 .select(fn.YEAR(Post.created_at).alias('year'), fn.COUNT(Post.id))
                 .group_by(fn.YEAR(Post.created_at)))

# CASE expressions
priority_case = Case(None, [
    (Post.is_urgent == True, 'High'),
    (Post.created_at > datetime.now() - timedelta(hours=1), 'Medium'),
], 'Low')

posts_with_priority = Post.select(Post.title, priority_case.alias('priority'))

# CAST expressions
user_ages_as_text = User.select(Cast(User.age, 'TEXT'))

# Raw SQL
custom_function = SQL('MY_CUSTOM_FUNCTION(?)', User.id)
results = User.select(User.username, custom_function.alias('custom_value'))

Subqueries and CTEs

Subquery construction and Common Table Expressions (CTEs) for complex hierarchical and analytical queries.

class Select:
    def alias(self, alias):
        """
        Create aliased subquery.
        
        Parameters:
        - alias (str): Alias name for subquery
        
        Returns:
        Select: Aliased query usable in other queries
        """
    
    def cte(self, name, recursive=False, columns=None):
        """
        Create Common Table Expression.
        
        Parameters:
        - name (str): CTE name
        - recursive (bool): Enable recursive CTE
        - columns (list): Column names for CTE
        
        Returns:
        CTE: Common Table Expression
        """

def WITH(*ctes):
    """
    Create query with CTEs.
    
    Parameters:
    - *ctes: Common Table Expressions
    
    Returns:
    Query with attached CTEs
    """

Usage examples:

# Subqueries
subquery = User.select(User.id).where(User.age > 18)
posts = Post.select().where(Post.author.in_(subquery))

# Correlated subqueries
subquery = (Post
            .select(fn.COUNT(Post.id))
            .where(Post.author == User.id))
users_with_post_count = User.select(User.username, subquery.alias('post_count'))

# Common Table Expressions (CTEs)
base_cte = User.select(User.id, User.username, User.manager_id).cte('employees')

recursive_cte = (base_cte
                 .select_from(base_cte.c.id, base_cte.c.username, base_cte.c.manager_id)
                 .union_all(
                     User.select(User.id, User.username, User.manager_id)
                     .join(base_cte, on=(User.manager_id == base_cte.c.id))
                 ))

with_cte_query = (WITH(recursive_cte)
                  .select_from(recursive_cte.c.username)
                  .order_by(recursive_cte.c.username))

Bulk Operations

Efficient bulk insert, update, and delete operations for handling large datasets with performance optimizations.

class Model:
    @classmethod
    def bulk_create(cls, model_list, batch_size=None):
        """
        Bulk create multiple instances.
        
        Parameters:
        - model_list (list): List of model instances
        - batch_size (int): Number of records per batch
        
        Returns:
        int: Number of created records
        """
    
    @classmethod
    def bulk_update(cls, model_list, fields, batch_size=None):
        """
        Bulk update multiple instances.
        
        Parameters:
        - model_list (list): List of model instances
        - fields (list): Fields to update
        - batch_size (int): Number of records per batch
        
        Returns:
        int: Number of updated records
        """
    
    @classmethod
    def insert_many(cls, rows, fields=None):
        """
        Insert multiple rows from data.
        
        Parameters:
        - rows (list): List of dictionaries or tuples
        - fields (list): Field names (for tuple data)
        
        Returns:
        Insert query
        """
    
    @classmethod
    def insert_from(cls, query, fields):
        """
        Insert from SELECT query results.
        
        Parameters:
        - query: SELECT query to insert from
        - fields (list): Target fields
        
        Returns:
        Insert query
        """
    
    @classmethod
    def replace_many(cls, rows, fields=None):
        """
        Replace multiple rows (INSERT OR REPLACE).
        
        Parameters:
        - rows (list): List of dictionaries or tuples
        - fields (list): Field names (for tuple data)
        
        Returns:
        Insert query with REPLACE
        """

def chunked(iterable, n):
    """
    Split iterable into chunks of size n.
    
    Parameters:
    - iterable: Data to chunk
    - n (int): Chunk size
    
    Yields:
    Chunks of the iterable
    """

Usage examples:

from peewee import *

# Bulk create from instances
users = [
    User(username=f'user{i}', email=f'user{i}@example.com')
    for i in range(1000)
]
User.bulk_create(users, batch_size=100)

# Insert many from data
user_data = [
    {'username': 'john', 'email': 'john@example.com'},
    {'username': 'jane', 'email': 'jane@example.com'},
    # ... more data
]
User.insert_many(user_data).execute()

# Insert from tuples
user_tuples = [
    ('john', 'john@example.com'),
    ('jane', 'jane@example.com'),
]
User.insert_many(user_tuples, fields=[User.username, User.email]).execute()

# Insert from query
active_users = User.select().where(User.is_active == True)
ArchivedUser.insert_from(active_users, [ArchivedUser.username, ArchivedUser.email]).execute()

# Bulk update
users = User.select().where(User.last_login.is_null())
for user in users:
    user.is_active = False

User.bulk_update(users, fields=[User.is_active], batch_size=100)

# Chunked processing
all_users = User.select()
for user_batch in chunked(all_users, 100):
    process_user_batch(user_batch)

Prefetching and N+1 Prevention

Optimization techniques for efficiently loading related objects and preventing N+1 query problems.

def prefetch(query, *subqueries, **kwargs):
    """
    Prefetch related objects to avoid N+1 queries.
    
    Parameters:
    - query: Base query to prefetch for
    - *subqueries: Related object queries to prefetch
    - prefetch_type: Prefetch strategy (WHERE, JOIN)
    
    Returns:
    Generator of objects with prefetched relations
    """

class Model:
    @classmethod
    def select(cls, *fields):
        def join(self, dest, join_type='INNER', on=None):
            """Join related tables in single query."""
        
        def switch(self, dest=None):
            """Switch context for additional joins."""

Usage examples:

# N+1 problem (BAD - multiple queries)
users = User.select()
for user in users:
    print(f"{user.username} has {len(user.posts)} posts")  # One query per user

# Solution 1: Prefetch
users_with_posts = prefetch(User.select(), Post.select())
for user in users_with_posts:
    print(f"{user.username} has {len(user.posts)} posts")  # Only 2 queries total

# Solution 2: Join with aggregation
users = (User
         .select(User.username, fn.COUNT(Post.id).alias('post_count'))
         .join(Post, JOIN.LEFT_OUTER)
         .group_by(User.username))

for user in users:
    print(f"{user.username} has {user.post_count} posts")  # Single query

# Complex prefetching
query = (User
         .select()
         .where(User.is_active == True))

posts_query = (Post
               .select()
               .where(Post.is_published == True))

comments_query = (Comment
                  .select()
                  .where(Comment.is_approved == True))

# Prefetch users with their posts and comments
users_with_data = prefetch(query, posts_query, comments_query)
for user in users_with_data:
    print(f"User: {user.username}")
    for post in user.posts:
        print(f"  Post: {post.title}")
        for comment in post.comments:
            print(f"    Comment: {comment.content}")

Query Utility Classes and Functions

Core utility classes and functions for query construction, value handling, and SQL expression building.

class AsIs:
    """
    Pass-through value wrapper without escaping.
    
    Parameters:
    - value: Value to pass through without modification
    """
    def __init__(self, value): ...

class Value:
    """
    Literal value wrapper for SQL expressions.
    
    Parameters:
    - value: Literal value to wrap
    - converter: Optional value converter function
    """
    def __init__(self, value, converter=None): ...

class ValuesList:
    """
    VALUES list for bulk insert operations.
    
    Parameters:
    - values (list): List of value tuples
    """
    def __init__(self, values): ...

class Column:
    """
    Column reference for queries and expressions.
    
    Parameters:
    - source: Source table or alias
    - name (str): Column name
    """
    def __init__(self, source, name): ...

class Table:
    """
    Table reference for queries and joins.
    
    Parameters:
    - name (str): Table name
    - alias (str): Table alias
    """
    def __init__(self, name, alias=None): ...

class Window:
    """
    SQL window function definitions.
    
    Parameters:
    - partition_by: Fields to partition by
    - order_by: Fields to order by
    - start: Window frame start
    - end: Window frame end
    """
    def __init__(self, partition_by=None, order_by=None, start=None, end=None): ...

class Context:
    """
    Query execution context management.
    """
    def __init__(self, **kwargs): ...

class QualifiedNames:
    """
    Qualified column name wrapper for complex queries.
    
    Parameters:
    - source: Source model or table
    """
    def __init__(self, source): ...

class DQ:
    """
    Dynamic query helper for programmatic query construction.
    
    Parameters:
    - **kwargs: Dynamic query conditions
    """
    def __init__(self, **kwargs): ...

def Check(constraint):
    """
    Check constraint helper function.
    
    Parameters:
    - constraint (str): SQL constraint expression
    
    Returns:
    Check constraint object
    """

def Asc(node, collation=None, nulls=None):
    """
    Create ascending order expression.
    
    Parameters:
    - node: Field or expression to order by
    - collation (str): Collation name
    - nulls (str): NULL handling ('FIRST' or 'LAST')
    
    Returns:
    Ordering: Ascending order expression
    """

def Desc(node, collation=None, nulls=None):
    """
    Create descending order expression.
    
    Parameters:
    - node: Field or expression to order by
    - collation (str): Collation name  
    - nulls (str): NULL handling ('FIRST' or 'LAST')
    
    Returns:
    Ordering: Descending order expression
    """

def chunked(iterable, n):
    """
    Utility to chunk iterables into smaller batches.
    
    Parameters:
    - iterable: Data to chunk
    - n (int): Chunk size
    
    Yields:
    Chunks of the iterable
    """

def Tuple(*args):
    """
    Create tuple expression for multi-column operations.
    
    Parameters:
    - *args: Values or expressions to include in tuple
    
    Returns:
    EnclosedNodeList: Tuple expression
    """

class Tuple:
    """Alias for tuple expressions using EnclosedNodeList."""

class Value:
    """
    Literal value wrapper for SQL expressions.
    
    Parameters:
    - value: Literal value to wrap
    - converter: Optional value converter function
    - unpack (bool): Unpack multi-value types like lists
    """
    def __init__(self, value, converter=None, unpack=True): ...

class ValuesList:
    """
    VALUES clause for table-valued expressions.
    
    Parameters:
    - values (list): List of value tuples  
    - columns (list): Column names for values
    - alias (str): Table alias
    """
    def __init__(self, values, columns=None, alias=None): ...
    
    def columns(self, *names):
        """Set column names for VALUES table."""

class Window:
    """
    SQL window function specification.
    
    Parameters:
    - partition_by: Fields to partition by
    - order_by: Fields to order by  
    - start: Frame start boundary
    - end: Frame end boundary
    - frame_type (str): Frame type ('ROWS', 'RANGE', 'GROUPS')
    - extends: Base window to extend
    - exclude: Frame exclusion type
    - alias (str): Window alias
    """
    # Frame boundaries
    CURRENT_ROW = SQL('CURRENT ROW')
    GROUP = SQL('GROUP') 
    TIES = SQL('TIES')
    NO_OTHERS = SQL('NO OTHERS')
    
    # Frame types
    GROUPS = 'GROUPS'
    RANGE = 'RANGE'
    ROWS = 'ROWS'
    
    def __init__(self, partition_by=None, order_by=None, start=None, end=None,
                 frame_type=None, extends=None, exclude=None, alias=None): ...

class ModelIndex:
    """
    Database index definition for models.
    
    Parameters:
    - model: Model class to create index for
    - fields (list): Fields to include in index
    - unique (bool): Create unique index
    - safe (bool): Don't raise error if index exists
    - where: Partial index condition
    - using (str): Index method
    - name (str): Custom index name
    """
    def __init__(self, model, fields, unique=False, safe=True, where=None,
                 using=None, name=None): ...

Constants and Operators

Important constants and operator definitions for query building and special operations.

class OP:
    """
    Operator constants for query expressions.
    """
    EQ = 'EQ'           # Equals (=)
    LT = 'LT'           # Less than (<)
    LTE = 'LTE'         # Less than or equal (<=)
    GT = 'GT'           # Greater than (>)
    GTE = 'GTE'         # Greater than or equal (>=)
    NE = 'NE'           # Not equal (!=)
    IN = 'IN'           # IN clause
    IS = 'IS'           # IS clause
    LIKE = 'LIKE'       # LIKE pattern matching
    ILIKE = 'ILIKE'     # Case-insensitive LIKE
    REGEXP = 'REGEXP'   # Regular expression matching
    IREGEXP = 'IREGEXP' # Case-insensitive regex
    BETWEEN = 'BETWEEN' # BETWEEN clause
    ADD = 'ADD'         # Addition (+)
    SUB = 'SUB'         # Subtraction (-)
    MUL = 'MUL'         # Multiplication (*)
    DIV = 'DIV'         # Division (/)
    BIN_AND = 'BIN_AND' # Binary AND (&)
    BIN_OR = 'BIN_OR'   # Binary OR (|)
    XOR = 'XOR'         # Exclusive OR
    MOD = 'MOD'         # Modulo (%)
    LSHIFT = 'LSHIFT'   # Left shift (<<)
    RSHIFT = 'RSHIFT'   # Right shift (>>)
    AND = 'AND'         # Logical AND
    OR = 'OR'           # Logical OR
    NOT = 'NOT'         # Logical NOT
    CONTAINS = 'CONTAINS'     # String contains
    STARTSWITH = 'STARTSWITH' # String starts with
    ENDSWITH = 'ENDSWITH'     # String ends with
    CONCAT = 'CONCAT'         # String concatenation
    NULL = 'NULL'             # NULL check

class JOIN:
    """
    Join type constants.
    """
    INNER = 'INNER'           # INNER JOIN
    LEFT = 'LEFT'             # LEFT JOIN  
    LEFT_OUTER = 'LEFT OUTER' # LEFT OUTER JOIN
    RIGHT = 'RIGHT'           # RIGHT JOIN
    RIGHT_OUTER = 'RIGHT OUTER' # RIGHT OUTER JOIN
    FULL = 'FULL'             # FULL JOIN
    FULL_OUTER = 'FULL OUTER' # FULL OUTER JOIN
    CROSS = 'CROSS'           # CROSS JOIN

EXCLUDED = 'EXCLUDED'
"""Reference to excluded values in UPSERT operations (PostgreSQL)."""

DJANGO_MAP = {
    'AutoField': 'AutoField',
    'BigAutoField': 'BigAutoField', 
    'BigIntegerField': 'BigIntegerField',
    'BinaryField': 'BlobField',
    'BooleanField': 'BooleanField',
    'CharField': 'CharField',
    'DateField': 'DateField',
    'DateTimeField': 'DateTimeField',
    'DecimalField': 'DecimalField',
    'DurationField': 'IntegerField',
    'EmailField': 'CharField',
    'FileField': 'CharField',
    'FilePathField': 'CharField',
    'FloatField': 'FloatField',
    'ImageField': 'CharField',
    'IntegerField': 'IntegerField',
    'GenericIPAddressField': 'CharField',
    'NullBooleanField': 'BooleanField',
    'PositiveIntegerField': 'IntegerField',
    'PositiveSmallIntegerField': 'SmallIntegerField',
    'SlugField': 'CharField',
    'SmallIntegerField': 'SmallIntegerField',
    'TextField': 'TextField',
    'TimeField': 'TimeField',
    'URLField': 'CharField',
    'UUIDField': 'UUIDField',
}
"""Mapping of Django ORM field types to Peewee field types."""

PREFETCH_TYPE = 'PREFETCH_TYPE'
"""Prefetch strategy constants for optimization."""

Usage examples:

from peewee import *

# Using operators explicitly
query = User.select().where(User.age.op(OP.BETWEEN)(18, 65))

# Join types
query = (User
         .select()
         .join(Post, JOIN.LEFT_OUTER)
         .where(User.is_active == True))

# Value wrappers
raw_value = AsIs("NOW()")  # Pass SQL function without escaping
literal_value = Value(42)   # Literal value

# Column references for complex queries
user_table = Table('user', 'u')
username_col = Column(user_table, 'username')

# Check constraints
class User(Model):
    age = IntegerField(constraints=[Check('age >= 0')])
    
# Dynamic queries
conditions = {'is_active': True, 'age__gt': 18}
dq = DQ(**conditions)
users = User.select().where(dq)

# EXCLUDED for UPSERT (PostgreSQL)
query = (User
         .insert(username='john', email='john@example.com')
         .on_conflict(
             conflict_target=[User.username],
             update={User.email: EXCLUDED.email}))

Install with Tessl CLI