Convert mysql bit column to Python boolean

[jhoelzl]

Hello!

When i select a bit column from a mysql table (with a raw query), the value for 1 looks like b'\x01'.

In mysqlclient package i can adjust the mapping between mysql and Python data types by the conv parameter:

from MySQLdb import converters

def convert_bit_to_bool(bit):
    """
    Convert Mysql bit to Python bool
    When Mysql bit is NULL, then use Python None instead
    :param bit:
    :return: converted_value
    """

    if bit:
        converted_value = bool(ord(bit))
    else:
        converted_value = None

    return converted_value

data_conversation = converters.conversions
data_conversation[MySQLdb.constants.FIELD_TYPE.BIT] = convert_bit_to_bool

connection = MySQLdb.connect(
            host='xxx',
            port=1234,
            user=xxx,
            passwd=xxx,
            db='xxx',
            charset='utf8',
            conv=data_conversation,
        )

How can i achieve this with the databases library?
Thanks!

[ljluestc]

1. Define the Custom Converter:

from sqlalchemy import create_engine, MetaData
from sqlalchemy.dialects.mysql import BIT
from sqlalchemy.types import TypeDecorator, BOOLEAN

class BitBool(TypeDecorator):
    impl = BOOLEAN

    def process_bind_param(self, value, dialect):
        if value is not None:
            return int(value)
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            return bool(value)
        return value

2. Create the Database Connection:

import databases
import sqlalchemy

DATABASE_URL = "mysql://user:password@host:port/database"

database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()

3. Define a Sample Table with the Custom Type:

from sqlalchemy import Table, Column, Integer, String

example_table = Table(
    "example",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(length=50)),
    Column("is_active", BitBool),
)

4. Execute Queries:

import asyncio

async def main():
    # Connect to the database
    await database.connect()
    
    # Insert a sample row
    query = example_table.insert().values(id=1, name="Sample", is_active=True)
    await database.execute(query)
    
    # Retrieve the row
    query = example_table.select().where(example_table.c.id == 1)
    row = await database.fetch_one(query)
    print(row)  # This will show 'is_active' as a boolean
    
    # Disconnect from the database
    await database.disconnect()

# Run the main function
asyncio.run(main())

[jhoelzl]

Thank you very much!