Data Type Handling

def escape_string(value):
    """
    Escape a string value for safe SQL inclusion.
    
    Parameters:
    - value (str): String value to escape
    
    Returns:
    str: Escaped string with quotes
    """

def escape_dict(val, charset):
    """
    Escape all values in a dictionary.
    
    Parameters:
    - val (dict): Dictionary with values to escape
    - charset (str): Character set for encoding
    
    Returns:
    dict: Dictionary with escaped values
    """

def escape_sequence(val, charset):
    """
    Escape a sequence of values for SQL tuple.
    
    Parameters:
    - val (sequence): Sequence of values to escape
    - charset (str): Character set for encoding
    
    Returns:
    str: Escaped sequence as SQL tuple string
    """

def Binary(x):
    """
    Create a binary data object for MySQL BLOB fields.
    
    Parameters:
    - x: Data to convert to binary
    
    Returns:
    bytes: Binary data object
    """

Date = date
"""Alias for datetime.date - represents MySQL DATE type."""

Time = time  
"""Alias for datetime.time - represents MySQL TIME type."""

Timestamp = datetime
"""Alias for datetime.datetime - represents MySQL DATETIME/TIMESTAMP types."""

def DateFromTicks(ticks):
    """
    Create a date object from Unix timestamp.
    
    Parameters:
    - ticks (float): Unix timestamp
    
    Returns:
    date: Date object from timestamp
    """

def TimeFromTicks(ticks):
    """
    Create a time object from Unix timestamp.
    
    Parameters:
    - ticks (float): Unix timestamp
    
    Returns:
    time: Time object from timestamp
    """

def TimestampFromTicks(ticks):
    """
    Create a datetime object from Unix timestamp.
    
    Parameters:
    - ticks (float): Unix timestamp
    
    Returns:
    datetime: Datetime object from timestamp
    """

# Field type constants from cymysql.constants.FIELD_TYPE
DECIMAL = 0
TINY = 1
SHORT = 2
LONG = 3
FLOAT = 4
DOUBLE = 5
NULL = 6
TIMESTAMP = 7
LONGLONG = 8
INT24 = 9
DATE = 10
TIME = 11
DATETIME = 12
YEAR = 13
NEWDATE = 14
VARCHAR = 15
BIT = 16
VECTOR = 242
JSON = 245
NEWDECIMAL = 246
ENUM = 247
SET = 248
TINY_BLOB = 249
MEDIUM_BLOB = 250
LONG_BLOB = 251
BLOB = 252
VAR_STRING = 253
STRING = 254
GEOMETRY = 255

# Type aliases
CHAR = TINY
INTERVAL = ENUM

class DBAPISet(frozenset):
    """
    Special frozenset for DB-API type comparison.
    
    Supports 'in' operator and equality comparison with individual types.
    """
    
    def __eq__(self, other):
        """Check if type is in this set."""
        
    def __ne__(self, other):
        """Check if type is not in this set."""

STRING = DBAPISet([FIELD_TYPE.ENUM, FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
"""Set of string-like field types."""

BINARY = DBAPISet([FIELD_TYPE.BLOB, FIELD_TYPE.LONG_BLOB, 
                  FIELD_TYPE.MEDIUM_BLOB, FIELD_TYPE.TINY_BLOB])
"""Set of binary field types."""

NUMBER = DBAPISet([FIELD_TYPE.DECIMAL, FIELD_TYPE.DOUBLE, FIELD_TYPE.FLOAT,
                  FIELD_TYPE.INT24, FIELD_TYPE.LONG, FIELD_TYPE.LONGLONG,
                  FIELD_TYPE.TINY, FIELD_TYPE.YEAR])
"""Set of numeric field types."""

DATE = DBAPISet([FIELD_TYPE.DATE, FIELD_TYPE.NEWDATE])
"""Set of date field types."""

TIME = DBAPISet([FIELD_TYPE.TIME])
"""Set of time field types."""

TIMESTAMP = DBAPISet([FIELD_TYPE.TIMESTAMP, FIELD_TYPE.DATETIME])
"""Set of timestamp/datetime field types."""

DATETIME = TIMESTAMP
"""Alias for TIMESTAMP type set."""

ROWID = DBAPISet()
"""Empty set - MySQL doesn't have ROWID."""

# Built-in converter functions (from cymysql.converters module)
def escape_bool(value):
    """Convert Python bool to MySQL boolean representation."""

def escape_int(value):
    """Convert Python int to MySQL integer representation."""  

def escape_float(value):
    """Convert Python float to MySQL float representation."""

def escape_bytes(value):
    """Convert Python bytes to MySQL binary representation."""

import cymysql
from datetime import date, datetime, time

conn = cymysql.connect(host='localhost', user='root', db='test')
cursor = conn.cursor()

# Insert date/time data
current_date = date.today()
current_time = datetime.now()

cursor.execute(
    "INSERT INTO events (event_date, event_timestamp) VALUES (%s, %s)",
    (current_date, current_time)
)

# Query date/time data
cursor.execute("SELECT event_date, event_timestamp FROM events")
for row in cursor.fetchall():
    event_date, event_timestamp = row
    print(f"Date: {event_date}, Timestamp: {event_timestamp}")

conn.commit()
cursor.close()
conn.close()

import cymysql

conn = cymysql.connect(host='localhost', user='root', db='test')
cursor = conn.cursor()

# Store binary data
with open('image.jpg', 'rb') as f:
    image_data = f.read()

binary_data = cymysql.Binary(image_data)
cursor.execute(
    "INSERT INTO images (name, data) VALUES (%s, %s)",
    ('photo.jpg', binary_data)
)

# Retrieve binary data
cursor.execute("SELECT name, data FROM images WHERE name = %s", ('photo.jpg',))
name, data = cursor.fetchone()

with open('retrieved_image.jpg', 'wb') as f:
    f.write(data)

conn.commit()
cursor.close()
conn.close()

import cymysql
from cymysql import FIELD_TYPE, STRING, NUMBER, DATE

conn = cymysql.connect(host='localhost', user='root', db='test')
cursor = conn.cursor()

cursor.execute("SELECT id, name, salary, hire_date FROM employees LIMIT 1")

# Check column types
for i, desc in enumerate(cursor.description):
    column_name = desc[0]
    field_type = desc[1]
    
    print(f"Column: {column_name}")
    
    if field_type in STRING:
        print("  Type: String")
    elif field_type in NUMBER:
        print("  Type: Number")
    elif field_type in DATE:
        print("  Type: Date")
    elif field_type == FIELD_TYPE.TIMESTAMP:
        print("  Type: Timestamp")
    else:
        print(f"  Type: Unknown ({field_type})")

cursor.close()
conn.close()

import cymysql

# Manual escaping (not recommended - use parameterized queries instead)
unsafe_input = "'; DROP TABLE users; --"
safe_string = cymysql.escape_string(unsafe_input)
print(f"Escaped: {safe_string}")

# Proper parameterized query (recommended)
conn = cymysql.connect(host='localhost', user='root', db='test')
cursor = conn.cursor()

user_input = "O'Reilly"
cursor.execute("SELECT * FROM users WHERE name = %s", (user_input,))
# CyMySQL automatically escapes the parameter safely

cursor.close()
conn.close()

import cymysql
import json

conn = cymysql.connect(host='localhost', user='root', db='test')
cursor = conn.cursor()

# Store JSON data
user_profile = {
    'preferences': {'theme': 'dark', 'language': 'en'},
    'settings': {'notifications': True, 'email_updates': False}
}

cursor.execute(
    "INSERT INTO users (name, profile) VALUES (%s, %s)",
    ('john_doe', json.dumps(user_profile))
)

# Retrieve and parse JSON data
cursor.execute("SELECT name, profile FROM users WHERE name = %s", ('john_doe',))
name, profile_json = cursor.fetchone()

profile = json.loads(profile_json) if profile_json else {}
print(f"User: {name}")
print(f"Theme: {profile.get('preferences', {}).get('theme')}")

conn.commit()
cursor.close()
conn.close()

import cymysql

conn = cymysql.connect(host='localhost', user='root', db='test')
cursor = conn.cursor()

# Insert NULL values
cursor.execute(
    "INSERT INTO users (name, email, phone) VALUES (%s, %s, %s)",
    ('John Doe', 'john@example.com', None)  # None becomes NULL
)

# Query with NULL handling
cursor.execute("SELECT name, email, phone FROM users")
for row in cursor.fetchall():
    name, email, phone = row
    phone_display = phone if phone is not None else 'No phone'
    print(f"{name}: {email}, {phone_display}")

conn.commit()
cursor.close()
conn.close()

import cymysql
import decimal
from datetime import datetime

# Custom converter for high-precision decimals
def convert_decimal(value):
    return decimal.Decimal(value.decode('utf-8'))

# Custom converter for timestamps
def convert_timestamp(value):
    return datetime.strptime(value.decode('utf-8'), '%Y-%m-%d %H:%M:%S')

# Note: Custom converters would be registered with connection
# This is an advanced feature for specialized type handling

import cymysql
from datetime import datetime, date

conn = cymysql.connect(host='localhost', user='root', db='test')
cursor = conn.cursor()

# Mixed data types in bulk insert
employees = [
    ('Alice Johnson', 75000.50, date(2020, 1, 15), True),
    ('Bob Smith', 82000.00, date(2019, 6, 10), True),
    ('Carol Davis', 68000.25, date(2021, 3, 8), False)
]

cursor.executemany(
    "INSERT INTO employees (name, salary, hire_date, active) VALUES (%s, %s, %s, %s)",
    employees
)

print(f"Inserted {cursor.rowcount} employees")

conn.commit()
cursor.close()
conn.close()