<commit_before>from os import path
from textwrap import dedent
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.dialects import sqlite, mysql
from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey

def test_lazy_connecting(tmpdir):
    db_file = path.join(tmpdir.strpath, 'test.db')
    engine = create_engine('sqlite:///%s' % db_file) # Database URL
    assert isinstance(engine, sqlalchemy.engine.Engine)
    assert not path.exists(db_file) # lazy connecting

    conn = engine.connect()
    assert isinstance(conn, sqlalchemy.engine.Connection)
    assert path.exists(db_file)

def test_create_tables_sqlite():
    metadata = MetaData()
    users = Table('user', metadata,
            Column('id', Integer, primary_key=True),
            Column('name', String),
            Column('fullname', String),
    )
    addresses = Table('address', metadata,
            Column('id', Integer, primary_key=True),
            Column('user_id', None, ForeignKey('user.id')),
            Column('email', String, nullable=False),
    )

    assert create_table_sql(users, sqlite) == dedent("""\
        CREATE TABLE user (
        	id INTEGER NOT NULL, 
        	name VARCHAR, 
        	fullname VARCHAR, 
        	PRIMARY KEY (id)
        )""")

    assert create_table_sql(addresses, sqlite) == dedent("""\
        CREATE TABLE address (
        	id INTEGER NOT NULL, 
        	user_id INTEGER, 
        	email VARCHAR NOT NULL, 
        	PRIMARY KEY (id), 
        	FOREIGN KEY(user_id) REFERENCES user (id)
        )""")

def test_create_tables_mysql():
    # CompileError: (in table 'user', column 'name'): VARCHAR requires a length on dialect mysql
    metadata = MetaData()
    users = Table('user', metadata,
            Column('id', Integer, primary_key=True),
            Column('name', String(20)),
            Column('fullname', String(50)),
    )
    addresses = Table('address', metadata,
            Column('id', Integer, primary_key=True),
            Column('user_id', None, ForeignKey('user.id')),
            Column('email', String(100), nullable=False),
    )

    assert create_table_sql(users, mysql) == dedent("""\
        CREATE TABLE user (
        	id INTEGER NOT NULL AUTO_INCREMENT, 
        	name VARCHAR(20), 
        	fullname VARCHAR(50), 
        	PRIMARY KEY (id)
        )""")

    assert create_table_sql(addresses, mysql) == dedent("""\
        CREATE TABLE address (
        	id INTEGER NOT NULL AUTO_INCREMENT, 
        	user_id INTEGER, 
        	email VARCHAR(100) NOT NULL, 
        	PRIMARY KEY (id), 
        	FOREIGN KEY(user_id) REFERENCES user (id)
        )""")

def create_table_sql(table, dialect_module):
    from sqlalchemy.schema import CreateTable
    return str(CreateTable(table).compile(dialect=dialect_module.dialect())).strip()

<commit_msg>[sqlalchemy] Create tables<commit_after>from os import path
from textwrap import dedent
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.dialects import sqlite, mysql
from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey

def test_lazy_connecting(tmpdir):
    db_file = path.join(tmpdir.strpath, 'test.db')
    engine = create_engine('sqlite:///%s' % db_file) # Database URL
    assert isinstance(engine, sqlalchemy.engine.Engine)
    assert not path.exists(db_file) # lazy connecting

    conn = engine.connect()
    assert isinstance(conn, sqlalchemy.engine.Connection)
    assert path.exists(db_file)

def test_define_tables_sqlite():
    metadata = MetaData()
    users = Table('user', metadata,
            Column('id', Integer, primary_key=True),
            Column('name', String),
            Column('fullname', String),
    )
    addresses = Table('address', metadata,
            Column('id', Integer, primary_key=True),
            Column('user_id', None, ForeignKey('user.id')),
            Column('email', String, nullable=False),
    )

    assert create_table_sql(users, sqlite) == dedent("""\
        CREATE TABLE user (
        	id INTEGER NOT NULL, 
        	name VARCHAR, 
        	fullname VARCHAR, 
        	PRIMARY KEY (id)
        )""")

    assert create_table_sql(addresses, sqlite) == dedent("""\
        CREATE TABLE address (
        	id INTEGER NOT NULL, 
        	user_id INTEGER, 
        	email VARCHAR NOT NULL, 
        	PRIMARY KEY (id), 
        	FOREIGN KEY(user_id) REFERENCES user (id)
        )""")

def test_define_tables_mysql():
    # CompileError: (in table 'user', column 'name'): VARCHAR requires a length on dialect mysql
    metadata = MetaData()
    users = Table('user', metadata,
            Column('id', Integer, primary_key=True),
            Column('name', String(20)),
            Column('fullname', String(50)),
    )
    addresses = Table('address', metadata,
            Column('id', Integer, primary_key=True),
            Column('user_id', None, ForeignKey('user.id')),
            Column('email', String(100), nullable=False),
    )

    assert create_table_sql(users, mysql) == dedent("""\
        CREATE TABLE user (
        	id INTEGER NOT NULL AUTO_INCREMENT, 
        	name VARCHAR(20), 
        	fullname VARCHAR(50), 
        	PRIMARY KEY (id)
        )""")

    assert create_table_sql(addresses, mysql) == dedent("""\
        CREATE TABLE address (
        	id INTEGER NOT NULL AUTO_INCREMENT, 
        	user_id INTEGER, 
        	email VARCHAR(100) NOT NULL, 
        	PRIMARY KEY (id), 
        	FOREIGN KEY(user_id) REFERENCES user (id)
        )""")

def create_table_sql(table, dialect_module):
    from sqlalchemy.schema import CreateTable
    return str(CreateTable(table).compile(dialect=dialect_module.dialect())).strip()

def test_create_tables(tmpdir):
    db_file = path.join(tmpdir.strpath, 'test.db')

    engine = create_engine('sqlite:///%s' % db_file)
    metadata = MetaData()
    users = Table('user', metadata, Column('id', Integer, primary_key=True))

    metadata.create_all(engine)
    assert sqlite3(db_file, '.schema'), dedent("""\
        CREATE TABLE user (
        	id INTEGER NOT NULL,
        	PRIMARY KEY (id)
        );""")

def sqlite3(db_file, command):
    import subprocess
    return subprocess.check_output(['sqlite3', db_file, command])


{'fraction_non_alphanumeric': 0.11760925449871465, 'fraction_numerical': 0.004980719794344473, 'mean_word_length': 2.781895504252734, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '900505', 'n_tokens_mistral': 1967, 'n_tokens_neox': 1814, 'n_words': 472}