File mysql-to-sqlite3-2.4.1.obscpio of Package python-mysql-to-sqlite3
07070100000000000081A4000000000000000000000001682E58C100002011000000000000000000000000000000000000002300000000mysql-to-sqlite3-2.4.1/.bandit.yml### This config may optionally select a subset of tests to run or skip by
### filling out the 'tests' and 'skips' lists given below. If no tests are
### specified for inclusion then it is assumed all tests are desired. The skips
### set will remove specific tests from the include set. This can be controlled
### using the -t/-s CLI options. Note that the same test ID should not appear
### in both 'tests' and 'skips', this would be nonsensical and is detected by
### Bandit at runtime.
# Available tests:
# B101 : assert_used
# B102 : exec_used
# B103 : set_bad_file_permissions
# B104 : hardcoded_bind_all_interfaces
# B105 : hardcoded_password_string
# B106 : hardcoded_password_funcarg
# B107 : hardcoded_password_default
# B108 : hardcoded_tmp_directory
# B110 : try_except_pass
# B112 : try_except_continue
# B201 : flask_debug_true
# B301 : pickle
# B302 : marshal
# B303 : md5
# B304 : ciphers
# B305 : cipher_modes
# B306 : mktemp_q
# B307 : eval
# B308 : mark_safe
# B309 : httpsconnection
# B310 : urllib_urlopen
# B311 : random
# B312 : telnetlib
# B313 : xml_bad_cElementTree
# B314 : xml_bad_ElementTree
# B315 : xml_bad_expatreader
# B316 : xml_bad_expatbuilder
# B317 : xml_bad_sax
# B318 : xml_bad_minidom
# B319 : xml_bad_pulldom
# B320 : xml_bad_etree
# B321 : ftplib
# B322 : input
# B323 : unverified_context
# B324 : hashlib_new_insecure_functions
# B325 : tempnam
# B401 : import_telnetlib
# B402 : import_ftplib
# B403 : import_pickle
# B404 : import_subprocess
# B405 : import_xml_etree
# B406 : import_xml_sax
# B407 : import_xml_expat
# B408 : import_xml_minidom
# B409 : import_xml_pulldom
# B410 : import_lxml
# B411 : import_xmlrpclib
# B412 : import_httpoxy
# B413 : import_pycrypto
# B501 : request_with_no_cert_validation
# B502 : ssl_with_bad_version
# B503 : ssl_with_bad_defaults
# B504 : ssl_with_no_version
# B505 : weak_cryptographic_key
# B506 : yaml_load
# B507 : ssh_no_host_key_verification
# B601 : paramiko_calls
# B602 : subprocess_popen_with_shell_equals_true
# B603 : subprocess_without_shell_equals_true
# B604 : any_other_function_with_shell_equals_true
# B605 : start_process_with_a_shell
# B606 : start_process_with_no_shell
# B607 : start_process_with_partial_path
# B608 : hardcoded_sql_expressions
# B609 : linux_commands_wildcard_injection
# B610 : django_extra_used
# B611 : django_rawsql_used
# B701 : jinja2_autoescape_false
# B702 : use_of_mako_templates
# B703 : django_mark_safe
# (optional) list included test IDs here, eg '[B101, B406]':
tests:
# (optional) list skipped test IDs here, eg '[B101, B406]':
skips:
- B404
- B603
- B607
- B608
### (optional) plugin settings - some test plugins require configuration data
### that may be given here, per-plugin. All bandit test plugins have a built in
### set of sensible defaults and these will be used if no configuration is
### provided. It is not necessary to provide settings for every (or any) plugin
### if the defaults are acceptable.
any_other_function_with_shell_equals_true:
no_shell:
- os.execl
- os.execle
- os.execlp
- os.execlpe
- os.execv
- os.execve
- os.execvp
- os.execvpe
- os.spawnl
- os.spawnle
- os.spawnlp
- os.spawnlpe
- os.spawnv
- os.spawnve
- os.spawnvp
- os.spawnvpe
- os.startfile
shell:
- os.system
- os.popen
- os.popen2
- os.popen3
- os.popen4
- popen2.popen2
- popen2.popen3
- popen2.popen4
- popen2.Popen3
- popen2.Popen4
- commands.getoutput
- commands.getstatusoutput
subprocess:
- subprocess.Popen
- subprocess.call
- subprocess.check_call
- subprocess.check_output
- subprocess.run
hardcoded_tmp_directory:
tmp_dirs:
- /tmp
- /var/tmp
- /dev/shm
linux_commands_wildcard_injection:
no_shell:
- os.execl
- os.execle
- os.execlp
- os.execlpe
- os.execv
- os.execve
- os.execvp
- os.execvpe
- os.spawnl
- os.spawnle
- os.spawnlp
- os.spawnlpe
- os.spawnv
- os.spawnve
- os.spawnvp
- os.spawnvpe
- os.startfile
shell:
- os.system
- os.popen
- os.popen2
- os.popen3
- os.popen4
- popen2.popen2
- popen2.popen3
- popen2.popen4
- popen2.Popen3
- popen2.Popen4
- commands.getoutput
- commands.getstatusoutput
subprocess:
- subprocess.Popen
- subprocess.call
- subprocess.check_call
- subprocess.check_output
- subprocess.run
ssl_with_bad_defaults:
bad_protocol_versions:
- PROTOCOL_SSLv2
- SSLv2_METHOD
- SSLv23_METHOD
- PROTOCOL_SSLv3
- PROTOCOL_TLSv1
- SSLv3_METHOD
- TLSv1_METHOD
ssl_with_bad_version:
bad_protocol_versions:
- PROTOCOL_SSLv2
- SSLv2_METHOD
- SSLv23_METHOD
- PROTOCOL_SSLv3
- PROTOCOL_TLSv1
- SSLv3_METHOD
- TLSv1_METHOD
start_process_with_a_shell:
no_shell:
- os.execl
- os.execle
- os.execlp
- os.execlpe
- os.execv
- os.execve
- os.execvp
- os.execvpe
- os.spawnl
- os.spawnle
- os.spawnlp
- os.spawnlpe
- os.spawnv
- os.spawnve
- os.spawnvp
- os.spawnvpe
- os.startfile
shell:
- os.system
- os.popen
- os.popen2
- os.popen3
- os.popen4
- popen2.popen2
- popen2.popen3
- popen2.popen4
- popen2.Popen3
- popen2.Popen4
- commands.getoutput
- commands.getstatusoutput
subprocess:
- subprocess.Popen
- subprocess.call
- subprocess.check_call
- subprocess.check_output
- subprocess.run
start_process_with_no_shell:
no_shell:
- os.execl
- os.execle
- os.execlp
- os.execlpe
- os.execv
- os.execve
- os.execvp
- os.execvpe
- os.spawnl
- os.spawnle
- os.spawnlp
- os.spawnlpe
- os.spawnv
- os.spawnve
- os.spawnvp
- os.spawnvpe
- os.startfile
shell:
- os.system
- os.popen
- os.popen2
- os.popen3
- os.popen4
- popen2.popen2
- popen2.popen3
- popen2.popen4
- popen2.Popen3
- popen2.Popen4
- commands.getoutput
- commands.getstatusoutput
subprocess:
- subprocess.Popen
- subprocess.call
- subprocess.check_call
- subprocess.check_output
- subprocess.run
start_process_with_partial_path:
no_shell:
- os.execl
- os.execle
- os.execlp
- os.execlpe
- os.execv
- os.execve
- os.execvp
- os.execvpe
- os.spawnl
- os.spawnle
- os.spawnlp
- os.spawnlpe
- os.spawnv
- os.spawnve
- os.spawnvp
- os.spawnvpe
- os.startfile
shell:
- os.system
- os.popen
- os.popen2
- os.popen3
- os.popen4
- popen2.popen2
- popen2.popen3
- popen2.popen4
- popen2.Popen3
- popen2.Popen4
- commands.getoutput
- commands.getstatusoutput
subprocess:
- subprocess.Popen
- subprocess.call
- subprocess.check_call
- subprocess.check_output
- subprocess.run
subprocess_popen_with_shell_equals_true:
no_shell:
- os.execl
- os.execle
- os.execlp
- os.execlpe
- os.execv
- os.execve
- os.execvp
- os.execvpe
- os.spawnl
- os.spawnle
- os.spawnlp
- os.spawnlpe
- os.spawnv
- os.spawnve
- os.spawnvp
- os.spawnvpe
- os.startfile
shell:
- os.system
- os.popen
- os.popen2
- os.popen3
- os.popen4
- popen2.popen2
- popen2.popen3
- popen2.popen4
- popen2.Popen3
- popen2.Popen4
- commands.getoutput
- commands.getstatusoutput
subprocess:
- subprocess.Popen
- subprocess.call
- subprocess.check_call
- subprocess.check_output
- subprocess.run
subprocess_without_shell_equals_true:
no_shell:
- os.execl
- os.execle
- os.execlp
- os.execlpe
- os.execv
- os.execve
- os.execvp
- os.execvpe
- os.spawnl
- os.spawnle
- os.spawnlp
- os.spawnlpe
- os.spawnv
- os.spawnve
- os.spawnvp
- os.spawnvpe
- os.startfile
shell:
- os.system
- os.popen
- os.popen2
- os.popen3
- os.popen4
- popen2.popen2
- popen2.popen3
- popen2.popen4
- popen2.Popen3
- popen2.Popen4
- commands.getoutput
- commands.getstatusoutput
subprocess:
- subprocess.Popen
- subprocess.call
- subprocess.check_call
- subprocess.check_output
- subprocess.run
try_except_continue:
check_typed_exception: false
try_except_pass:
check_typed_exception: false
weak_cryptographic_key:
weak_key_size_dsa_high: 1024
weak_key_size_dsa_medium: 2048
weak_key_size_ec_high: 160
weak_key_size_ec_medium: 224
weak_key_size_rsa_high: 1024
weak_key_size_rsa_medium: 2048
07070100000001000081A4000000000000000000000001682E58C10000006D000000000000000000000000000000000000001F00000000mysql-to-sqlite3-2.4.1/.flake8[flake8]
ignore = I100,I201,I202,D203,D401,W503,E203,F401,F403,C901,E501
exclude = tests
max-line-length = 8807070100000002000041ED000000000000000000000002682E58C100000000000000000000000000000000000000000000001F00000000mysql-to-sqlite3-2.4.1/.github07070100000003000081A4000000000000000000000001682E58C100000038000000000000000000000000000000000000002B00000000mysql-to-sqlite3-2.4.1/.github/FUNDING.ymlgithub: techouse
custom: [ "https://paypal.me/ktusar" ]
07070100000004000041ED000000000000000000000002682E58C100000000000000000000000000000000000000000000002E00000000mysql-to-sqlite3-2.4.1/.github/ISSUE_TEMPLATE07070100000005000081A4000000000000000000000001682E58C1000002BF000000000000000000000000000000000000003C00000000mysql-to-sqlite3-2.4.1/.github/ISSUE_TEMPLATE/bug_report.md---
name: Bug report
about: Create a report to help us improve
title: ''
labels: bug
assignees: techouse
---
**Describe the bug**
A clear and concise description of what the bug is.
**Expected behaviour**
What you expected.
**Actual result**
What happened instead.
**System Information**
```bash
$ mysql2sqlite --version
```
```
<paste here>
```
This command is only available on v1.3.2 and greater. Otherwise, please provide some basic information about your system (Python version, operating system, etc.).
**Additional context**
Add any other context about the problem here.
In case of errors please run the same command with `--debug`. This option is only available on v1.4.12 or greater.
07070100000006000081A4000000000000000000000001682E58C100000262000000000000000000000000000000000000004100000000mysql-to-sqlite3-2.4.1/.github/ISSUE_TEMPLATE/feature_request.md---
name: Feature request
about: Suggest an idea for this project
title: ''
labels: enhancement
assignees: techouse
---
**Is your feature request related to a problem? Please describe.**
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
**Describe the solution you'd like**
A clear and concise description of what you want to happen.
**Describe alternatives you've considered**
A clear and concise description of any alternative solutions or features you've considered.
**Additional context**
Add any other context or screenshots about the feature request here.
07070100000007000081A4000000000000000000000001682E58C1000000CC000000000000000000000000000000000000002E00000000mysql-to-sqlite3-2.4.1/.github/dependabot.ymlversion: 2
updates:
- package-ecosystem: "pip"
directory: "/"
schedule:
interval: "weekly"
- package-ecosystem: "github-actions"
directory: "/"
schedule:
interval: "weekly"07070100000008000081A4000000000000000000000001682E58C100000532000000000000000000000000000000000000003800000000mysql-to-sqlite3-2.4.1/.github/pull_request_template.md# Pull Request Template
## Description
Please include a summary of the change and which issue is fixed. Please also include relevant motivation and context.
List any dependencies that are required for this change.
Fixes # (issue)
## Type of change
Please delete options that are not relevant.
- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update
## How Has This Been Tested?
Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also
list any relevant details for your test configuration
- [ ] Test A
- [ ] Test B
## Checklist:
- [ ] My code follows the style guidelines of this project
- [ ] I have performed a self-review of my own code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my feature works
- [ ] New and existing unit tests pass locally with my changes
- [ ] Any dependent changes have been merged and published in downstream modules07070100000009000041ED000000000000000000000002682E58C100000000000000000000000000000000000000000000002900000000mysql-to-sqlite3-2.4.1/.github/workflows0707010000000A000081A4000000000000000000000001682E58C100000666000000000000000000000000000000000000003400000000mysql-to-sqlite3-2.4.1/.github/workflows/docker.ymlname: Publish Docker image
on:
workflow_call:
defaults:
run:
shell: bash
jobs:
push_to_registry:
name: Push Docker image to Docker Hub
runs-on: ubuntu-latest
permissions:
packages: write
contents: read
environment:
name: docker
url: https://hub.docker.com/r/${{ vars.DOCKERHUB_REPOSITORY }}
steps:
- name: Check out the repo
uses: actions/checkout@v4
- name: Set up QEMU
uses: docker/setup-qemu-action@v3
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Docker meta
id: meta
uses: docker/metadata-action@v5
with:
images: |
${{ vars.DOCKERHUB_REPOSITORY }}
ghcr.io/${{ github.repository }}
tags: |
type=ref,event=branch
type=ref,event=pr
type=semver,pattern={{version}}
type=semver,pattern={{major}}.{{minor}}.{{patch}}
- name: Log in to Docker Hub
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: Log in to the Container registry
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Build and push
uses: docker/build-push-action@v6
with:
context: .
platforms: linux/amd64,linux/arm64
push: true
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
0707010000000B000081A4000000000000000000000001682E58C10000034B000000000000000000000000000000000000003200000000mysql-to-sqlite3-2.4.1/.github/workflows/docs.ymlname: Docs
on:
workflow_dispatch:
branches:
- main
workflow_call:
permissions:
contents: write
jobs:
docs:
name: "Docs"
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.x"
cache: "pip"
- name: Install package
run: pip install -e .
- name: Install Sphinx dependencies
working-directory: docs
run: pip install -r requirements.txt
- name: Sphinx build
working-directory: docs
run: sphinx-build . _build
- name: Deploy to GitHub Pages
uses: peaceiris/actions-gh-pages@v4
with:
publish_branch: gh-pages
github_token: ${{ secrets.GITHUB_TOKEN }}
publish_dir: docs/_build/
force_orphan: true0707010000000C000081A4000000000000000000000001682E58C100000DCA000000000000000000000000000000000000003500000000mysql-to-sqlite3-2.4.1/.github/workflows/publish.ymlname: Publish package
on:
push:
tags:
- 'v[0-9]+.[0-9]+.[0-9]+*'
defaults:
run:
shell: bash
permissions: read-all
jobs:
test:
uses: ./.github/workflows/test.yml
secrets: inherit
publish:
needs: test
name: "Publish"
runs-on: ubuntu-latest
environment:
name: pypi
url: https://pypi.org/p/mysql-to-sqlite3
permissions:
id-token: write
contents: write
steps:
- uses: actions/checkout@v4
- name: Compare package version with ref/tag
id: compare
run: |
set -e
VERSION=$(awk -F'"' '/__version__/ {print $2}' src/mysql_to_sqlite3/__init__.py)
TAG=${GITHUB_REF_NAME#v}
if [[ "$VERSION" != "$TAG" ]]; then
echo "Version in src/mysql_to_sqlite3/__version__.py ($VERSION) does not match tag ($TAG)"
exit 1
fi
echo "VERSION=$VERSION" >> $GITHUB_ENV
- name: Check CHANGELOG.md
id: check_changelog
run: |
set -e
if ! grep -q "# $VERSION" CHANGELOG.md; then
echo "CHANGELOG.md does not contain a section for $VERSION"
exit 1
fi
- name: Set up Python
id: setup_python
uses: actions/setup-python@v5
with:
python-version: "3.x"
- name: Install build dependencies
id: install_build_dependencies
run: |
set -e
python3 -m pip install --upgrade pip
pip install build setuptools wheel
- name: Build a binary wheel and a source tarball
id: build
run: |
set -e
python3 -m build --sdist --wheel --outdir dist/ .
- name: Publish distribution package to PyPI
id: publish
if: startsWith(github.ref, 'refs/tags')
uses: pypa/gh-action-pypi-publish@67339c736fd9354cd4f8cb0b744f2b82a74b5c70 # v1.12.3
- name: Install pyproject-parser
id: install_pyproject_parser
run: |
set -e
pip install pyproject-parser[cli]
- name: Read project name from pyproject.toml
id: read_project_name
run: |
set -e
NAME=$(pyproject-parser info project.name -r | tr -d '"')
echo "NAME=$NAME" >> $GITHUB_ENV
- name: Create tag-specific CHANGELOG
id: create_changelog
run: |
set -e
CHANGELOG_PATH=$RUNNER_TEMP/CHANGELOG.md
awk '/^#[[:space:]].*/ { if (count == 1) exit; count++; print } count == 1 && !/^#[[:space:]].*/ { print }' CHANGELOG.md | sed -e :a -e '/^\n*$/{$d;N;ba' -e '}' > $CHANGELOG_PATH
echo -en "\n[https://pypi.org/project/$NAME/$VERSION/](https://pypi.org/project/$NAME/$VERSION/)" >> $CHANGELOG_PATH
echo "CHANGELOG_PATH=$CHANGELOG_PATH" >> $GITHUB_ENV
- name: Github Release
id: github_release
uses: softprops/action-gh-release@v2
with:
name: ${{ env.VERSION }}
tag_name: ${{ github.ref }}
body_path: ${{ env.CHANGELOG_PATH }}
files: |
dist/*.whl
dist/*.tar.gz
- name: Cleanup
if: ${{ always() }}
run: |
rm -rf dist
rm -rf $CHANGELOG_PATH
docker:
needs: [ test, publish ]
permissions:
packages: write
contents: read
uses: ./.github/workflows/docker.yml
secrets: inherit
docs:
uses: ./.github/workflows/docs.yml
needs: [ test, publish ]
permissions:
contents: write
secrets: inherit0707010000000D000081A4000000000000000000000001682E58C100003E7C000000000000000000000000000000000000003200000000mysql-to-sqlite3-2.4.1/.github/workflows/test.ymlname: Test
on:
push:
branches:
- master
pull_request:
branches:
- master
workflow_call:
defaults:
run:
shell: bash
permissions: read-all
jobs:
analyze:
name: "Analyze"
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.x"
- name: Install dependencies
run: |
python3 -m pip install --upgrade pip
pip install -r requirements_dev.txt
- name: Run static analysis
run: tox -e linters
test:
name: "Test"
needs: analyze
runs-on: ubuntu-latest
strategy:
matrix:
include:
- toxenv: "python3.9"
db: "mariadb:5.5"
legacy_db: 1
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mariadb:5.5"
legacy_db: 1
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mariadb:5.5"
legacy_db: 1
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mariadb:5.5"
legacy_db: 1
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mariadb:5.5"
legacy_db: 1
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mariadb:10.0"
legacy_db: 1
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mariadb:10.0"
legacy_db: 1
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mariadb:10.0"
legacy_db: 1
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mariadb:10.0"
legacy_db: 1
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mariadb:10.0"
legacy_db: 1
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mariadb:10.1"
legacy_db: 1
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mariadb:10.1"
legacy_db: 1
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mariadb:10.1"
legacy_db: 1
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mariadb:10.1"
legacy_db: 1
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mariadb:10.1"
legacy_db: 1
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mariadb:10.2"
legacy_db: 0
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mariadb:10.2"
legacy_db: 0
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mariadb:10.2"
legacy_db: 0
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mariadb:10.2"
legacy_db: 0
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mariadb:10.2"
legacy_db: 0
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mariadb:10.3"
legacy_db: 0
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mariadb:10.3"
legacy_db: 0
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mariadb:10.3"
legacy_db: 0
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mariadb:10.3"
legacy_db: 0
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mariadb:10.3"
legacy_db: 0
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mariadb:10.4"
legacy_db: 0
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mariadb:10.4"
legacy_db: 0
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mariadb:10.4"
legacy_db: 0
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mariadb:10.4"
legacy_db: 0
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mariadb:10.4"
legacy_db: 0
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mariadb:10.5"
legacy_db: 0
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mariadb:10.5"
legacy_db: 0
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mariadb:10.5"
legacy_db: 0
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mariadb:10.5"
legacy_db: 0
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mariadb:10.5"
legacy_db: 0
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mariadb:10.6"
legacy_db: 0
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mariadb:10.6"
legacy_db: 0
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mariadb:10.6"
legacy_db: 0
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mariadb:10.6"
legacy_db: 0
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mariadb:10.6"
legacy_db: 0
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mariadb:10.11"
legacy_db: 0
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mariadb:10.11"
legacy_db: 0
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mariadb:10.11"
legacy_db: 0
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mariadb:10.11"
legacy_db: 0
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mariadb:10.11"
legacy_db: 0
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mariadb:11.4"
legacy_db: 0
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mariadb:11.4"
legacy_db: 0
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mariadb:11.4"
legacy_db: 0
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mariadb:11.4"
legacy_db: 0
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mariadb:11.4"
legacy_db: 0
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mysql:5.5"
legacy_db: 1
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mysql:5.5"
legacy_db: 1
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mysql:5.5"
legacy_db: 1
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mysql:5.5"
legacy_db: 1
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mysql:5.5"
legacy_db: 1
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mysql:5.6"
legacy_db: 1
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mysql:5.6"
legacy_db: 1
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mysql:5.6"
legacy_db: 1
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mysql:5.6"
legacy_db: 1
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mysql:5.6"
legacy_db: 1
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mysql:5.7"
legacy_db: 0
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mysql:5.7"
legacy_db: 0
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mysql:5.7"
legacy_db: 0
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mysql:5.7"
legacy_db: 0
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mysql:5.7"
legacy_db: 0
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mysql:8.0"
legacy_db: 0
experimental: false
py: "3.9"
- toxenv: "python3.10"
db: "mysql:8.0"
legacy_db: 0
experimental: false
py: "3.10"
- toxenv: "python3.11"
db: "mysql:8.0"
legacy_db: 0
experimental: false
py: "3.11"
- toxenv: "python3.12"
db: "mysql:8.0"
legacy_db: 0
experimental: false
py: "3.12"
- toxenv: "python3.13"
db: "mysql:8.0"
legacy_db: 0
experimental: false
py: "3.13"
- toxenv: "python3.9"
db: "mysql:8.4"
legacy_db: 0
experimental: true
py: "3.9"
- toxenv: "python3.10"
db: "mysql:8.4"
legacy_db: 0
experimental: true
py: "3.10"
- toxenv: "python3.11"
db: "mysql:8.4"
legacy_db: 0
experimental: true
py: "3.11"
- toxenv: "python3.12"
db: "mysql:8.4"
legacy_db: 0
experimental: true
py: "3.12"
- toxenv: "python3.13"
db: "mysql:8.4"
legacy_db: 0
experimental: true
py: "3.13"
continue-on-error: ${{ matrix.experimental }}
services:
mysql:
image: ${{ matrix.db }}
ports:
- 3306:3306
env:
MYSQL_ALLOW_EMPTY_PASSWORD: yes
options: >-
--name=mysqld
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.py }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.py }}
- uses: actions/cache@v4
with:
path: ~/.cache/pip
key: ${{ runner.os }}-pip-1
restore-keys: |
${{ runner.os }}-pip-
- name: Install dependencies
run: |
set -e
python -m pip install --upgrade pip
python -m pip install -U codecov tox-gh-actions
pip install -r requirements_dev.txt
- name: Set up MySQL
env:
DB: ${{ matrix.db }}
MYSQL_USER: tester
MYSQL_PASSWORD: testpass
MYSQL_DATABASE: test_db
MYSQL_HOST: 0.0.0.0
MYSQL_PORT: 3306
run: |
set -e
while :
do
sleep 1
mysql -h127.0.0.1 -uroot -e 'select version()' && break
done
case "$DB" in
'mysql:8.0'|'mysql:8.4')
mysql -h127.0.0.1 -uroot -e "SET GLOBAL local_infile=on"
docker cp mysqld:/var/lib/mysql/public_key.pem "${HOME}"
docker cp mysqld:/var/lib/mysql/ca.pem "${HOME}"
docker cp mysqld:/var/lib/mysql/server-cert.pem "${HOME}"
docker cp mysqld:/var/lib/mysql/client-key.pem "${HOME}"
docker cp mysqld:/var/lib/mysql/client-cert.pem "${HOME}"
;;
esac
USER_CREATION_COMMANDS=''
WITH_PLUGIN=''
if [ "$DB" == 'mysql:8.0' ]; then
WITH_PLUGIN='with mysql_native_password'
USER_CREATION_COMMANDS='
CREATE USER
user_sha256 IDENTIFIED WITH "sha256_password" BY "pass_sha256",
nopass_sha256 IDENTIFIED WITH "sha256_password",
user_caching_sha2 IDENTIFIED WITH "caching_sha2_password" BY "pass_caching_sha2",
nopass_caching_sha2 IDENTIFIED WITH "caching_sha2_password"
PASSWORD EXPIRE NEVER;
GRANT RELOAD ON *.* TO user_caching_sha2;'
elif [ "$DB" == 'mysql:8.4' ]; then
WITH_PLUGIN='with caching_sha2_password'
USER_CREATION_COMMANDS='
CREATE USER
user_caching_sha2 IDENTIFIED WITH "caching_sha2_password" BY "pass_caching_sha2",
nopass_caching_sha2 IDENTIFIED WITH "caching_sha2_password"
PASSWORD EXPIRE NEVER;
GRANT RELOAD ON *.* TO user_caching_sha2;'
fi
if [ ! -z "$USER_CREATION_COMMANDS" ]; then
mysql -uroot -h127.0.0.1 -e "$USER_CREATION_COMMANDS"
fi
mysql -h127.0.0.1 -uroot -e "create database $MYSQL_DATABASE DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
mysql -h127.0.0.1 -uroot -e "create user $MYSQL_USER identified $WITH_PLUGIN by '${MYSQL_PASSWORD}'; grant all on ${MYSQL_DATABASE}.* to ${MYSQL_USER};"
mysql -h127.0.0.1 -uroot -e "create user ${MYSQL_USER}@localhost identified $WITH_PLUGIN by '${MYSQL_PASSWORD}'; grant all on ${MYSQL_DATABASE}.* to ${MYSQL_USER}@localhost;"
- name: Create db_credentials.json
env:
MYSQL_USER: tester
MYSQL_PASSWORD: testpass
MYSQL_DATABASE: test_db
MYSQL_HOST: 0.0.0.0
MYSQL_PORT: 3306
run: |
set -e
jq -n \
--arg mysql_user "$MYSQL_USER" \
--arg mysql_password "$MYSQL_PASSWORD" \
--arg mysql_database "$MYSQL_DATABASE" \
--arg mysql_host "$MYSQL_HOST" \
--arg mysql_port $MYSQL_PORT \
'$ARGS.named' > tests/db_credentials.json
- name: Test with tox
env:
LEGACY_DB: ${{ matrix.legacy_db }}
run: tox
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v5
continue-on-error: true
with:
token: ${{ secrets.CODECOV_TOKEN }}
slug: techouse/mysql-to-sqlite3
files: ./coverage.xml
env_vars: OS,PYTHON
verbose: true
- name: Cleanup
if: ${{ always() }}
run: |
rm -rf tests/db_credentials.json
0707010000000E000081A4000000000000000000000001682E58C1000005AE000000000000000000000000000000000000002200000000mysql-to-sqlite3-2.4.1/.gitignore# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
pip-wheel-metadata/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# IDE specific
.idea
# macOS specific
.DS_Store
# Potential leftovers
tests/db_credentials.json
log.txt
0707010000000F000081A4000000000000000000000001682E58C100001C32000000000000000000000000000000000000002400000000mysql-to-sqlite3-2.4.1/CHANGELOG.md# 2.4.1
* [FIX] fix passwordless login
# 2.4.0
* [CHORE] drop support for Python 3.8
# 2.3.0
* [FEAT] add MySQL 8.4 and MariaDB 11.4 support
# 2.2.2
* [FIX] use `dateutil.parse` to parse SQLite dates
# 2.2.1
* [FIX] fix transferring composite primary keys when AUTOINCREMENT present
# 2.2.0
* [FEAT] add --without-tables option
# 2.1.12
* [CHORE] update MySQL Connector/Python to 8.4.0
* [CHORE] add Sphinx documentation
# 2.1.11
* [CHORE] migrate package from flat layout to src layout
# 2.1.10
* [FEAT] add support for AUTOINCREMENT
# 2.1.9
* [FIX] pin MySQL Connector/Python to 8.3.0
# 2.1.8
* [FIX] ensure index names do not collide with table names
# 2.1.7
* [FIX] use more precise foreign key constraints
# 2.1.6
* [FEAT] build both linux/amd64 and linux/arm64 Docker images
# 2.1.5
* [CHORE] fix Docker package publishing from Github Workflow
# 2.1.4
* [FIX] fix invalid column_type error message
# 2.1.3
* [CHORE] maintenance release to publish first containerized release
# 2.1.2
* [FIX] throw more comprehensive error messages when translating column types
# 2.1.1
* [CHORE] add support for Python 3.12
* [CHORE] bump minimum version of MySQL Connector/Python to 8.2.0
# 2.1.0
* [CHORE] drop support for Python 3.7
# 2.0.3
* [FIX] import MySQLConnectionAbstract instead of concrete implementations
# 2.0.2
* [FIX] properly import CMySQLConnection
# 2.0.1
* [FEAT] add support for MySQL character set introducers in DEFAULT clause
# 2.0.0
* [CHORE] drop support for Python 2.7, 3.5 and 3.6
* [CHORE] migrate pytest.ini configuration into pyproject.toml
* [CHORE] migrate from setuptools to hatch / hatchling
* [CHORE] update dependencies
* [CHORE] add types
* [CHORE] add types to tests
* [CHORE] update dependencies
* [CHORE] use f-strings where appropriate
# 1.4.18
* [CHORE] update dependencies
* [CHORE] use [black](https://github.com/psf/black) and [isort](https://github.com/PyCQA/isort) in tox linters
# 1.4.17
* [CHORE] migrate from setup.py to pyproject.toml
* [CHORE] update the publishing workflow
# 1.4.16
* [CHORE] add MariaDB 10.11 CI tests
* [CHORE] add Python 3.11 support
# 1.4.15
* [FIX] fix BLOB default value
* [CHORE] remove CI tests for Python 3.5, 3.6, add tests for Python 3.11
# 1.4.14
* [FIX] pin mysql-connector-python to <8.0.30
* [CHORE] update CI actions/checkout to v3
* [CHORE] update CI actions/setup-python to v4
* [CHORE] update CI actions/cache to v3
* [CHORE] update CI github/codeql-action/init to v2
* [CHORE] update CI github/codeql-action/analyze to v2
# 1.4.13
* [FEAT] add option to exclude specific MySQL tables
* [CHORE] update CI codecov/codecov-action to v2
# 1.4.12
* [FIX] fix SQLite convert_date converter
* [CHORE] update tests
# 1.4.11
* [FIX] pin python-slugify to <6.0.0
# 1.4.10
* [FEAT] add feature to transfer tables without any data (DDL only)
# 1.4.9
* [CHORE] add Python 3.10 support
* [CHORE] add Python 3.10 tests
# 1.4.8
* [FEAT] transfer JSON columns as JSON
# 1.4.7
* [CHORE] add experimental tests for Python 3.10-dev
* [CHORE] add tests for MariaDB 10.6
# 1.4.6
* [FIX] pin Click to <8.0
# 1.4.5
* [FEAT] add -K, --prefix-indices CLI option to prefix indices with table names. This used to be the default behavior
until now. To keep the old behavior simply use this CLI option.
# 1.4.4
* [FEAT] add --limit-rows CLI option
* [FEAT] add --collation CLI option to specify SQLite collation sequence
# 1.4.3
* [FIX] pin python-tabulate to <0.8.6 for Python 3.4 or less
* [FIX] pin python-slugify to <5.0.0 for Python 3.5 or less
* [FIX] pin Click to 7.x for Python 3.5 or less
# 1.4.2
* [FIX] fix default column value not getting converted
# 1.4.1
* [FIX] get table list error when Click package is 8.0+
# 1.4.0
* [FEAT] add password prompt. This changes the default behavior of -p
* [FEAT] add option to disable MySQL connection encryption
* [FEAT] add non-chunked progress bar
* [FIX] pin mysql-connector-python to <8.0.24 for Python 3.5 or lower
* [FIX] require sqlalchemy <1.4.0 to make compatible with sqlalchemy-utils
# 1.3.8
* [FIX] some MySQL integer column definitions result in TEXT fields in sqlite3
* [FIX] fix CI tests
# 1.3.7
* [CHORE] transition from Travis CI to GitHub Actions
# 1.3.6
* [FIX] Fix Python 3.9 tests
# 1.3.5
* [FIX] add IF NOT EXISTS to the CREATE INDEX SQL command
* [CHORE] add Python 3.9 CI tests
# 1.3.4
* [FEAT] add --quiet option
# 1.3.3
* [FIX] test for mysql client more gracefully
# 1.3.2
* [FEAT] simpler access to the debug version info using the --version switch
* [FEAT] add debug_info module to be used in bug reports
* [CHORE] remove PyPy and PyPy3 CI tests
* [CHORE] add tabulate to development dependencies
* [CHORE] use pytest fixture fom Faker 4.1.0 in Python 3 tests
* [CHORE] omit debug_info.py in coverage reports
# 1.3.1
* [FIX] fix information_schema issue introduced with MySQL 8.0.21
* [FIX] fix MySQL 8 bug where column types would sometimes be returned as bytes instead of strings
* [FIX] sqlalchemy-utils dropped Python 2.7 support in v0.36.7
* [CHORE] use MySQL Client instead of PyMySQL in tests
* [CHORE] add MySQL version output to CI tests
* [CHORE] add Python 3.9 to the CI tests
* [CHORE] add MariaDB 10.5 to the CI tests
* [CHORE] remove Python 2.7 from allowed CI test failures
* [CHORE] use Ubuntu Bionic instead of Ubuntu Xenial in CI tests
* [CHORE] use Ubuntu Xenial only for MariaDB 10.4 CI tests
* [CHORE] test legacy databases in CI tests
# 1.3.0
* [FEAT] add option to transfer only specific tables using -t
* [CHORE] add tests for transferring only certain tables
# 1.2.11
* [FIX] duplicate foreign keys
# 1.2.10
* [FIX] properly escape SQLite index names
* [FIX] fix SQLite global index name scoping
* [CHORE] test the successful transfer of an unorthodox table name
* [CHORE] test the successful transfer of indices with same names
# 1.2.9
* [FIX] differentiate better between MySQL and SQLite errors
* [CHORE] add Python 3.8 and 3.8-dev test build
# 1.2.8
* [CHORE] add support for Python 3.8
* [CHORE] update mysql-connector-python to a minimum version of 8.0.18 to support Python 3.8
* [CHORE] update development dependencies
* [CHORE] add [bandit](https://github.com/PyCQA/bandit) tests
# 1.2.7
* [FEAT] transfer unique indices
* [FIX] improve index transport
* [CHORE] test transfer of indices
# 1.2.6
* [CHORE] include tests in the PyPI package
# 1.2.5
* [FEAT] transfer foreign keys
* [CHORE] removed duplicate import in test database models
# 1.2.4
* [CHORE] reformat MySQLtoSQLite constructor
* [CHORE] reformat translator function
* [CHORE] add more tests
# 1.2.3
* [CHORE] add more tests
# 1.2.2
* [CHORE] refactor package
* [CHORE] fix CI tests
* [CHORE] add linter rules
# 1.2.1
* [FEAT] add Python 2.7 support
# 1.2.0
* [CHORE] add CI tests
* [CHORE] achieve 100% test coverage
# 1.1.2
* [FIX] fix error of transferring tables without primary keys
* [FIX] fix error of transferring empty tables
# 1.1.1
* [FEAT] add option to use MySQLCursorBuffered cursors
* [FEAT] add MySQL port
* [FEAT] update --help hints
* [FIX] fix slugify import
* [FIX] cursor error
# 1.1.0
* [FEAT] add VACUUM option
# 1.0.0
Initial commit
07070100000010000081A4000000000000000000000001682E58C100001589000000000000000000000000000000000000002A00000000mysql-to-sqlite3-2.4.1/CODE-OF-CONDUCT.md# Contributor Covenant Code of Conduct
## Our Pledge
We as members, contributors, and leaders pledge to make participation in our
community a harassment-free experience for everyone, regardless of age, body
size, visible or invisible disability, ethnicity, sex characteristics, gender
identity and expression, level of experience, education, socio-economic status,
nationality, personal appearance, race, caste, color, religion, or sexual identity
and orientation.
We pledge to act and interact in ways that contribute to an open, welcoming,
diverse, inclusive, and healthy community.
## Our Standards
Examples of behavior that contributes to a positive environment for our
community include:
* Demonstrating empathy and kindness toward other people
* Being respectful of differing opinions, viewpoints, and experiences
* Giving and gracefully accepting constructive feedback
* Accepting responsibility and apologizing to those affected by our mistakes,
and learning from the experience
* Focusing on what is best not just for us as individuals, but for the
overall community
Examples of unacceptable behavior include:
* The use of sexualized language or imagery, and sexual attention or
advances of any kind
* Trolling, insulting or derogatory comments, and personal or political attacks
* Public or private harassment
* Publishing others' private information, such as a physical or email
address, without their explicit permission
* Other conduct which could reasonably be considered inappropriate in a
professional setting
## Enforcement Responsibilities
Community leaders are responsible for clarifying and enforcing our standards of
acceptable behavior and will take appropriate and fair corrective action in
response to any behavior that they deem inappropriate, threatening, offensive,
or harmful.
Community leaders have the right and responsibility to remove, edit, or reject
comments, commits, code, wiki edits, issues, and other contributions that are
not aligned to this Code of Conduct, and will communicate reasons for moderation
decisions when appropriate.
## Scope
This Code of Conduct applies within all community spaces, and also applies when
an individual is officially representing the community in public spaces.
Examples of representing our community include using an official e-mail address,
posting via an official social media account, or acting as an appointed
representative at an online or offline event.
## Enforcement
Instances of abusive, harassing, or otherwise unacceptable behavior may be
reported to the community leaders responsible for enforcement at
[techouse@gmail.com](mailto:techouse@gmail.com).
All complaints will be reviewed and investigated promptly and fairly.
All community leaders are obligated to respect the privacy and security of the
reporter of any incident.
## Enforcement Guidelines
Community leaders will follow these Community Impact Guidelines in determining
the consequences for any action they deem in violation of this Code of Conduct:
### 1. Correction
**Community Impact**: Use of inappropriate language or other behavior deemed
unprofessional or unwelcome in the community.
**Consequence**: A private, written warning from community leaders, providing
clarity around the nature of the violation and an explanation of why the
behavior was inappropriate. A public apology may be requested.
### 2. Warning
**Community Impact**: A violation through a single incident or series
of actions.
**Consequence**: A warning with consequences for continued behavior. No
interaction with the people involved, including unsolicited interaction with
those enforcing the Code of Conduct, for a specified period of time. This
includes avoiding interactions in community spaces as well as external channels
like social media. Violating these terms may lead to a temporary or
permanent ban.
### 3. Temporary Ban
**Community Impact**: A serious violation of community standards, including
sustained inappropriate behavior.
**Consequence**: A temporary ban from any sort of interaction or public
communication with the community for a specified period of time. No public or
private interaction with the people involved, including unsolicited interaction
with those enforcing the Code of Conduct, is allowed during this period.
Violating these terms may lead to a permanent ban.
### 4. Permanent Ban
**Community Impact**: Demonstrating a pattern of violation of community
standards, including sustained inappropriate behavior, harassment of an
individual, or aggression toward or disparagement of classes of individuals.
**Consequence**: A permanent ban from any sort of public interaction within
the community.
## Attribution
This Code of Conduct is adapted from the [Contributor Covenant][homepage],
version 2.1, available at
[https://www.contributor-covenant.org/version/2/1/code_of_conduct.html][v2.1].
Community Impact Guidelines were inspired by
[Mozilla's code of conduct enforcement ladder][Mozilla CoC].
For answers to common questions about this code of conduct, see the FAQ at
[https://www.contributor-covenant.org/faq][FAQ]. Translations are available
at [https://www.contributor-covenant.org/translations][translations].
[homepage]: https://www.contributor-covenant.org
[v2.1]: https://www.contributor-covenant.org/version/2/1/code_of_conduct.html
[Mozilla CoC]: https://github.com/mozilla/diversity
[FAQ]: https://www.contributor-covenant.org/faq
[translations]: https://www.contributor-covenant.org/translations
07070100000011000081A4000000000000000000000001682E58C100000BE0000000000000000000000000000000000000002700000000mysql-to-sqlite3-2.4.1/CONTRIBUTING.md# Contributing
I greatly appreciate your interest in reading this message, as this project requires volunteer developers to assist
in developing and maintaining it.
Before making any changes to this repository, please first discuss the proposed modifications with the repository owners
through an issue, email, or any other appropriate communication channel.
Please be aware that a [code of conduct](CODE-OF-CONDUCT.md) is in place, and should be adhered to during all
interactions related to the project.
## Python version support
Ensuring backward compatibility is an imperative requirement.
Currently, the tool supports Python versions 3.9, 3.10, 3.11, 3.12, and 3.13.
## MySQL version support
This tool is intended to fully support MySQL versions 5.5, 5.6, 5.7, and 8.0, including major forks like MariaDB.
We should prioritize and be dedicated to maintaining compatibility with these versions for a smooth user experience.
## Testing
As this project/tool involves the critical process of transferring data between different database types, it is of
utmost importance to ensure thorough testing. Please remember to write tests for any new code you create, utilizing the
[pytest](https://docs.pytest.org/en/latest/) framework for all test cases.
### Running the test suite
In order to run the test suite run these commands using a Docker MySQL image.
**Requires a running Docker instance!**
```bash
git clone https://github.com/techouse/mysql-to-sqlite3
cd mysql-to-sqlite3
python3 -m venv env
source env/bin/activate
pip install -e .
pip install -r requirements_dev.txt
tox
```
## Submitting changes
To contribute to this project, please submit a
new [pull request](https://github.com/techouse/mysql-to-sqlite3/pull/new/master) and provide a clear list of your
modifications. For guidance on creating pull requests, you can refer
to [this resource](http://help.github.com/pull-requests/).
When sending a pull request, we highly appreciate the inclusion of [pytest](https://docs.pytest.org/en/latest/) tests,
as we strive to enhance our test coverage. Following our coding conventions is essential, and it would be ideal if you
ensure that each commit focuses on a single feature.
For commits, please write clear log messages. While concise one-line messages are suitable for small changes, more
substantial modifications should follow a format similar to the example below:
```bash
git commit -m "A brief summary of the commit
>
> A paragraph describing what changed and its impact."
```
## Coding standards
It is essential to prioritize code readability and conciseness. To achieve this, we recommend
using [Black](https://github.com/psf/black) for code formatting.
Once your work is deemed complete, it is advisable to run the following command:
```bash
tox -e flake8,linters
```
This command executes various linters and checkers to identify any potential issues or inconsistencies in your code. By
following these guidelines, you can ensure a high-quality codebase.
Thanks,
Klemen Tusar07070100000012000081A4000000000000000000000001682E58C1000000C7000000000000000000000000000000000000002200000000mysql-to-sqlite3-2.4.1/DockerfileFROM python:3.12-alpine
LABEL maintainer="https://github.com/techouse"
RUN pip install --no-cache-dir --upgrade pip && \
pip install --no-cache-dir mysql-to-sqlite3
ENTRYPOINT ["mysql2sqlite"]07070100000013000081A4000000000000000000000001682E58C10000042D000000000000000000000000000000000000001F00000000mysql-to-sqlite3-2.4.1/LICENSEMIT License
Copyright (c) 2025 Klemen Tusar
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
07070100000014000081A4000000000000000000000001682E58C100001BD2000000000000000000000000000000000000002100000000mysql-to-sqlite3-2.4.1/README.md[](https://pypi.org/project/mysql-to-sqlite3/)
[](https://pypistats.org/packages/mysql-to-sqlite3)
[](https://pypi.org/project/mysql-to-sqlite3/)
[](https://img.shields.io/static/v1?label=MySQL&message=5.5+|+5.6+|+5.7+|+8.0+|+8.4&color=2b5d80)
[](https://img.shields.io/static/v1?label=MariaDB&message=5.5|+10.0+|+10.1+|+10.2+|+10.3+|+10.4+|+10.5|+11.4&color=C0765A)
[](https://github.com/techouse/mysql-to-sqlite3/blob/master/LICENSE)
[](CODE-OF-CONDUCT.md)
[](https://pypi.org/project/sqlite3-to-mysql/)
[](https://github.com/ambv/black)
[](https://www.codacy.com/manual/techouse/mysql-to-sqlite3?utm_source=github.com&utm_medium=referral&utm_content=techouse/mysql-to-sqlite3&utm_campaign=Badge_Grade)
[](https://github.com/techouse/mysql-to-sqlite3/actions/workflows/test.yml)
[](https://github.com/techouse/mysql-to-sqlite3/actions/workflows/codeql-analysis.yml)
[](https://github.com/techouse/mysql-to-sqlite3/actions/workflows/publish.yml)
[](https://codecov.io/gh/techouse/mysql-to-sqlite3)
[](https://github.com/sponsors/techouse)
[](https://github.com/techouse/mysql-to-sqlite3/stargazers)
# MySQL to SQLite3
#### A simple Python tool to transfer data from MySQL to SQLite 3.
### How to run
```bash
pip install mysql-to-sqlite3
mysql2sqlite --help
```
### Usage
```
Usage: mysql2sqlite [OPTIONS]
Options:
-f, --sqlite-file PATH SQLite3 database file [required]
-d, --mysql-database TEXT MySQL database name [required]
-u, --mysql-user TEXT MySQL user [required]
-p, --prompt-mysql-password Prompt for MySQL password
--mysql-password TEXT MySQL password
-t, --mysql-tables TUPLE Transfer only these specific tables (space
separated table names). Implies --without-
foreign-keys which inhibits the transfer of
foreign keys. Can not be used together with
--exclude-mysql-tables.
-e, --exclude-mysql-tables TUPLE
Transfer all tables except these specific
tables (space separated table names).
Implies --without-foreign-keys which
inhibits the transfer of foreign keys. Can
not be used together with --mysql-tables.
-L, --limit-rows INTEGER Transfer only a limited number of rows from
each table.
-C, --collation [BINARY|NOCASE|RTRIM]
Create datatypes of TEXT affinity using a
specified collation sequence. [default:
BINARY]
-K, --prefix-indices Prefix indices with their corresponding
tables. This ensures that their names remain
unique across the SQLite database.
-X, --without-foreign-keys Do not transfer foreign keys.
-Z, --without-tables Do not transfer tables, data only.
-W, --without-data Do not transfer table data, DDL only.
-h, --mysql-host TEXT MySQL host. Defaults to localhost.
-P, --mysql-port INTEGER MySQL port. Defaults to 3306.
--mysql-charset TEXT MySQL database and table character set
[default: utf8mb4]
--mysql-collation TEXT MySQL database and table collation
-S, --skip-ssl Disable MySQL connection encryption.
-c, --chunk INTEGER Chunk reading/writing SQL records
-l, --log-file PATH Log file
--json-as-text Transfer JSON columns as TEXT.
-V, --vacuum Use the VACUUM command to rebuild the SQLite
database file, repacking it into a minimal
amount of disk space
--use-buffered-cursors Use MySQLCursorBuffered for reading the
MySQL database. This can be useful in
situations where multiple queries, with
small result sets, need to be combined or
computed with each other.
-q, --quiet Quiet. Display only errors.
--debug Debug mode. Will throw exceptions.
--version Show the version and exit.
--help Show this message and exit.
```
#### Docker
If you don't want to install the tool on your system, you can use the Docker image instead.
```bash
docker run -it \
--workdir $(pwd) \
--volume $(pwd):$(pwd) \
--rm ghcr.io/techouse/mysql-to-sqlite3:latest \
--sqlite-file baz.db \
--mysql-user foo \
--mysql-password bar \
--mysql-database baz \
--mysql-host host.docker.internal
```
This will mount your host current working directory (pwd) inside the Docker container as the current working directory.
Any files Docker would write to the current working directory are written to the host directory where you did docker
run. Note that you have to also use a
[special hostname](https://docs.docker.com/desktop/networking/#use-cases-and-workarounds-for-all-platforms) `host.docker.internal`
to access your host machine from inside the Docker container.
#### Homebrew
If you're on macOS, you can install the tool using [Homebrew](https://brew.sh/).
```bash
brew tap techouse/mysql-to-sqlite3
brew install mysql-to-sqlite3
mysql2sqlite --help
```
07070100000015000081A4000000000000000000000001682E58C10000067C000000000000000000000000000000000000002300000000mysql-to-sqlite3-2.4.1/SECURITY.md# Security Policy
## Supported Versions
| Version | Supported |
|---------|--------------------|
| 2.1.x | :white_check_mark: |
| 2.0.x | :x: |
| 1.x.x | :x: |
## Reporting a Vulnerability
We take the security of our software seriously. If you believe you have found a security vulnerability, please report it
to us as described below.
**DO NOT CREATE A GITHUB ISSUE** reporting the vulnerability.
Instead, send an email to [techouse@gmail.com](mailto:techouse@gmail.com).
In the report, please include the following:
- Your name and affiliation (if any).
- A description of the technical details of the vulnerabilities. It is very important to let us know how we can
reproduce your findings.
- An explanation who can exploit this vulnerability, and what they gain when doing so -- write an attack scenario. This
will help us evaluate your submission quickly, especially if it is a complex or creative vulnerability.
- Whether this vulnerability is public or known to third parties. If it is, please provide details.
If you don’t get an acknowledgment from us or have heard nothing from us in a week, please contact us again.
We will send a response indicating the next steps in handling your report. We will keep you informed about the progress
towards a fix and full announcement.
We will not disclose your identity to the public without your permission. We strive to credit researchers in our
advisories when we release a fix, but only after getting your permission.
We appreciate your efforts to responsibly disclose your findings, and will make every effort to acknowledge your
contributions.07070100000016000041ED000000000000000000000002682E58C100000000000000000000000000000000000000000000001C00000000mysql-to-sqlite3-2.4.1/docs07070100000017000081A4000000000000000000000001682E58C10000027A000000000000000000000000000000000000002500000000mysql-to-sqlite3-2.4.1/docs/Makefile# Minimal makefile for Sphinx documentation
#
# You can set these variables from the command line, and also
# from the environment for the first two.
SPHINXOPTS ?=
SPHINXBUILD ?= sphinx-build
SOURCEDIR = .
BUILDDIR = _build
# Put it first so that "make" without argument is like "make help".
help:
@$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
.PHONY: help Makefile
# Catch-all target: route all unknown targets to Sphinx using the new
# "make mode" option. $(O) is meant as a shortcut for $(SPHINXOPTS).
%: Makefile
@$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
07070100000018000081A4000000000000000000000001682E58C100000F11000000000000000000000000000000000000002700000000mysql-to-sqlite3-2.4.1/docs/README.rstUsage
-----
Options
^^^^^^^
The command line options for the ``mysql2sqlite`` tool are as follows:
.. code-block:: bash
mysql2sqlite [OPTIONS]
Required Options
""""""""""""""""
- ``-f, --sqlite-file PATH``: SQLite3 database file. This option is required.
- ``-d, --mysql-database TEXT``: MySQL database name. This option is required.
- ``-u, --mysql-user TEXT``: MySQL user. This option is required.
Password Options
""""""""""""""""
- ``-p, --prompt-mysql-password``: Prompt for MySQL password.
- ``--mysql-password TEXT``: MySQL password.
Table Options
"""""""""""""
- ``-t, --mysql-tables TUPLE``: Transfer only these specific tables (space separated table names). Implies --without-foreign-keys which inhibits the transfer of foreign keys. Can not be used together with --exclude-mysql-tables.
- ``-e, --exclude-mysql-tables TUPLE``: Transfer all tables except these specific tables (space separated table names). Implies --without-foreign-keys which inhibits the transfer of foreign keys. Can not be used together with --mysql-tables.
Transfer Options
""""""""""""""""
- ``-L, --limit-rows INTEGER``: Transfer only a limited number of rows from each table.
- ``-C, --collation [BINARY|NOCASE|RTRIM]``: Create datatypes of TEXT affinity using a specified collation sequence. The default is BINARY.
- ``-K, --prefix-indices``: Prefix indices with their corresponding tables. This ensures that their names remain unique across the SQLite database.
- ``-X, --without-foreign-keys``: Do not transfer foreign keys.
- ``-Z, --without-tables``: Do not transfer tables, data only.
- ``-W, --without-data``: Do not transfer table data, DDL only.
Connection Options
""""""""""""""""""
- ``-h, --mysql-host TEXT``: MySQL host. Defaults to localhost.
- ``-P, --mysql-port INTEGER``: MySQL port. Defaults to 3306.
- ``--mysql-charset TEXT``: MySQL database and table character set. The default is utf8mb4.
- ``--mysql-collation TEXT``: MySQL database and table collation
- ``-S, --skip-ssl``: Disable MySQL connection encryption.
Other Options
"""""""""""""
- ``-c, --chunk INTEGER``: Chunk reading/writing SQL records.
- ``-l, --log-file PATH``: Log file.
- ``--json-as-text``: Transfer JSON columns as TEXT.
- ``-V, --vacuum``: Use the VACUUM command to rebuild the SQLite database file, repacking it into a minimal amount of disk space.
- ``--use-buffered-cursors``: Use MySQLCursorBuffered for reading the MySQL database. This can be useful in situations where multiple queries, with small result sets, need to be combined or computed with each other.
- ``-q, --quiet``: Quiet. Display only errors.
- ``--debug``: Debug mode. Will throw exceptions.
- ``--version``: Show the version and exit.
- ``--help``: Show this message and exit.
Docker
^^^^^^
If you don’t want to install the tool on your system, you can use the
Docker image instead.
.. code:: bash
docker run -it \
--workdir $(pwd) \
--volume $(pwd):$(pwd) \
--rm ghcr.io/techouse/mysql-to-sqlite3:latest \
--sqlite-file baz.db \
--mysql-user foo \
--mysql-password bar \
--mysql-database baz \
--mysql-host host.docker.internal
This will mount your host current working directory (pwd) inside the
Docker container as the current working directory. Any files Docker
would write to the current working directory are written to the host
directory where you did docker run. Note that you have to also use a
`special
hostname <https://docs.docker.com/desktop/networking/#use-cases-and-workarounds-for-all-platforms>`__
``host.docker.internal`` to access your host machine from inside the
Docker container.
Homebrew
^^^^^^^^
If you’re on macOS, you can install the tool using
`Homebrew <https://brew.sh/>`__.
.. code:: bash
brew tap techouse/mysql-to-sqlite3
brew install mysql-to-sqlite3
mysql2sqlite --help
07070100000019000081A4000000000000000000000001682E58C1000004C3000000000000000000000000000000000000002400000000mysql-to-sqlite3-2.4.1/docs/conf.py# Configuration file for the Sphinx documentation builder.
#
# For the full list of built-in configuration values, see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html
import os
import sys
sys.path.insert(0, os.path.abspath(".."))
from mysql_to_sqlite3 import __version__
# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
project = "mysql-to-sqlite3"
copyright = "2024, Klemen Tusar"
author = "Klemen Tusar"
release = __version__
# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
extensions = ["sphinx.ext.autodoc", "sphinx.ext.viewcode", "sphinx.ext.napoleon"]
napoleon_google_docstring = True
napoleon_include_init_with_doc = True
templates_path = ["_templates"]
exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
html_theme = "alabaster"
html_static_path = ["_static"]
0707010000001A000081A4000000000000000000000001682E58C100000E0A000000000000000000000000000000000000002600000000mysql-to-sqlite3-2.4.1/docs/index.rstMySQL to SQLite3
================
A simple Python tool to transfer data from MySQL to SQLite 3
|PyPI| |PyPI - Downloads| |PyPI - Python Version| |MySQL Support|
|MariaDB Support| |GitHub license| |Contributor Covenant| |PyPI - Format|
|Code style: black| |Codacy Badge| |Test Status| |CodeQL Status|
|Publish PyPI Package Status| |codecov| |GitHub Sponsors| |GitHub stars|
Installation
------------
.. code:: bash
pip install mysql-to-sqlite3
Basic Usage
-----------
.. code:: bash
mysql2sqlite -f path/to/foo.sqlite -d foo_db -u foo_user -p
.. toctree::
:maxdepth: 2
:caption: Contents:
README
modules
Indices and tables
==================
* :ref:`genindex`
* :ref:`modindex`
* :ref:`search`
.. |PyPI| image:: https://img.shields.io/pypi/v/mysql-to-sqlite3
:target: https://pypi.org/project/mysql-to-sqlite3/
.. |PyPI - Downloads| image:: https://img.shields.io/pypi/dm/mysql-to-sqlite3
:target: https://pypistats.org/packages/mysql-to-sqlite3
.. |PyPI - Python Version| image:: https://img.shields.io/pypi/pyversions/mysql-to-sqlite3
:target: https://pypi.org/project/mysql-to-sqlite3/
.. |MySQL Support| image:: https://img.shields.io/static/v1?label=MySQL&message=5.5+%7C+5.6+%7C+5.7+%7C+8.0&color=2b5d80
:target: https://img.shields.io/static/v1?label=MySQL&message=5.6+%7C+5.7+%7C+8.0&color=2b5d80
.. |MariaDB Support| image:: https://img.shields.io/static/v1?label=MariaDB&message=5.5+%7C+10.0+%7C+10.1+%7C+10.2+%7C+10.3+%7C+10.4+%7C+10.5+%7C+10.6%7C+10.11&color=C0765A
:target: https://img.shields.io/static/v1?label=MariaDB&message=10.0+%7C+10.1+%7C+10.2+%7C+10.3+%7C+10.4+%7C+10.5&color=C0765A
.. |GitHub license| image:: https://img.shields.io/github/license/techouse/mysql-to-sqlite3
:target: https://github.com/techouse/mysql-to-sqlite3/blob/master/LICENSE
.. |Contributor Covenant| image:: https://img.shields.io/badge/Contributor%20Covenant-2.1-4baaaa.svg
:target: CODE-OF-CONDUCT.md
.. |PyPI - Format| image:: https://img.shields.io/pypi/format/mysql-to-sqlite3
:target: https://pypi.org/project/sqlite3-to-mysql/
.. |Code style: black| image:: https://img.shields.io/badge/code%20style-black-000000.svg
:target: https://github.com/ambv/black
.. |Codacy Badge| image:: https://api.codacy.com/project/badge/Grade/64aae8e9599746d58d277852b35cc2bd
:target: https://www.codacy.com/manual/techouse/mysql-to-sqlite3?utm_source=github.com&utm_medium=referral&utm_content=techouse/mysql-to-sqlite3&utm_campaign=Badge_Grade
.. |Test Status| image:: https://github.com/techouse/mysql-to-sqlite3/actions/workflows/test.yml/badge.svg
:target: https://github.com/techouse/mysql-to-sqlite3/actions/workflows/test.yml
.. |CodeQL Status| image:: https://github.com/techouse/mysql-to-sqlite3/actions/workflows/codeql-analysis.yml/badge.svg
:target: https://github.com/techouse/mysql-to-sqlite3/actions/workflows/codeql-analysis.yml
.. |Publish PyPI Package Status| image:: https://github.com/techouse/mysql-to-sqlite3/actions/workflows/publish.yml/badge.svg
:target: https://github.com/techouse/mysql-to-sqlite3/actions/workflows/publish.yml
.. |codecov| image:: https://codecov.io/gh/techouse/mysql-to-sqlite3/branch/master/graph/badge.svg
:target: https://codecov.io/gh/techouse/mysql-to-sqlite3
.. |GitHub Sponsors| image:: https://img.shields.io/github/sponsors/techouse
:target: https://github.com/sponsors/techouse
.. |GitHub stars| image:: https://img.shields.io/github/stars/techouse/mysql-to-sqlite3.svg?style=social&label=Star&maxAge=2592000
:target: https://github.com/techouse/mysql-to-sqlite3/stargazers0707010000001B000081A4000000000000000000000001682E58C1000002FD000000000000000000000000000000000000002500000000mysql-to-sqlite3-2.4.1/docs/make.bat@ECHO OFF
pushd %~dp0
REM Command file for Sphinx documentation
if "%SPHINXBUILD%" == "" (
set SPHINXBUILD=sphinx-build
)
set SOURCEDIR=.
set BUILDDIR=_build
%SPHINXBUILD% >NUL 2>NUL
if errorlevel 9009 (
echo.
echo.The 'sphinx-build' command was not found. Make sure you have Sphinx
echo.installed, then set the SPHINXBUILD environment variable to point
echo.to the full path of the 'sphinx-build' executable. Alternatively you
echo.may add the Sphinx directory to PATH.
echo.
echo.If you don't have Sphinx installed, grab it from
echo.https://www.sphinx-doc.org/
exit /b 1
)
if "%1" == "" goto help
%SPHINXBUILD% -M %1 %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O%
goto end
:help
%SPHINXBUILD% -M help %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O%
:end
popd
0707010000001C000081A4000000000000000000000001682E58C100000055000000000000000000000000000000000000002800000000mysql-to-sqlite3-2.4.1/docs/modules.rstmysql_to_sqlite3
================
.. toctree::
:maxdepth: 4
mysql_to_sqlite3
0707010000001D000081A4000000000000000000000001682E58C10000057B000000000000000000000000000000000000003100000000mysql-to-sqlite3-2.4.1/docs/mysql_to_sqlite3.rstmysql\_to\_sqlite3 package
==========================
Submodules
----------
mysql\_to\_sqlite3.cli module
-----------------------------
.. automodule:: mysql_to_sqlite3.cli
:members:
:undoc-members:
:show-inheritance:
mysql\_to\_sqlite3.click\_utils module
--------------------------------------
.. automodule:: mysql_to_sqlite3.click_utils
:members:
:undoc-members:
:show-inheritance:
mysql\_to\_sqlite3.debug\_info module
-------------------------------------
.. automodule:: mysql_to_sqlite3.debug_info
:members:
:undoc-members:
:show-inheritance:
mysql\_to\_sqlite3.mysql\_utils module
--------------------------------------
.. automodule:: mysql_to_sqlite3.mysql_utils
:members:
:undoc-members:
:show-inheritance:
mysql\_to\_sqlite3.sqlite\_utils module
---------------------------------------
.. automodule:: mysql_to_sqlite3.sqlite_utils
:members:
:undoc-members:
:show-inheritance:
mysql\_to\_sqlite3.transporter module
-------------------------------------
.. automodule:: mysql_to_sqlite3.transporter
:members:
:undoc-members:
:show-inheritance:
mysql\_to\_sqlite3.types module
-------------------------------
.. automodule:: mysql_to_sqlite3.types
:members:
:undoc-members:
:show-inheritance:
Module contents
---------------
.. automodule:: mysql_to_sqlite3
:members:
:undoc-members:
:show-inheritance:
0707010000001E000081A4000000000000000000000001682E58C10000000D000000000000000000000000000000000000002D00000000mysql-to-sqlite3-2.4.1/docs/requirements.txtSphinx==8.2.30707010000001F000081A4000000000000000000000001682E58C100000D00000000000000000000000000000000000000002600000000mysql-to-sqlite3-2.4.1/pyproject.toml[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "mysql-to-sqlite3"
description = "A simple Python tool to transfer data from MySQL to SQLite 3"
readme = "README.md"
license = { text = "MIT" }
requires-python = ">=3.9"
authors = [
{ name = "Klemen Tusar", email = "techouse@gmail.com" },
]
keywords = [
"mysql",
"sqlite3",
"transfer",
"data",
"migrate",
"migration",
]
classifiers = [
"Development Status :: 5 - Production/Stable",
"Environment :: Console",
"Intended Audience :: End Users/Desktop",
"Intended Audience :: Developers",
"Intended Audience :: System Administrators",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: Implementation :: CPython",
"Topic :: Database",
]
dependencies = [
"Click>=8.1.3",
"mysql-connector-python>=9.0.0",
"pytimeparse2",
"python-dateutil>=2.9.0.post0",
"types_python_dateutil",
"python-slugify>=7.0.0",
"simplejson>=3.19.0",
"tqdm>=4.65.0",
"tabulate",
"typing_extensions",
]
dynamic = ["version"]
[project.urls]
Homepage = "https://techouse.github.io/mysql-to-sqlite3/"
Documentation = "https://techouse.github.io/mysql-to-sqlite3/"
Source = "https://github.com/techouse/mysql-to-sqlite3"
Changelog = "https://github.com/techouse/mysql-to-sqlite3/blob/master/CHANGELOG.md"
Sponsor = "https://github.com/sponsors/techouse"
PayPal = "https://paypal.me/ktusar"
[tool.hatch.version]
path = "src/mysql_to_sqlite3/__init__.py"
[tool.hatch.build.targets.sdist]
include = [
"src",
"tests",
"README.md",
"CHANGELOG.md",
"CODE-OF-CONDUCT.md",
"LICENSE",
"requirements_dev.txt",
]
[project.scripts]
mysql2sqlite = "mysql_to_sqlite3.cli:cli"
[tool.black]
line-length = 120
target-version = ["py39", "py310", "py311", "py312", "py313"]
include = '\.pyi?$'
exclude = '''
(
/(
\.eggs
| \.git
| \.hg
| \.mypy_cache
| \.tox
| \.venv
| _build
| buck-out
| build
| dist
| docs
)/
| foo.py
)
'''
[tool.isort]
line_length = 120
profile = "black"
lines_after_imports = 2
known_first_party = "mysql_to_sqlite3"
skip_gitignore = true
[tool.pytest.ini_options]
pythonpath = ["src"]
testpaths = ["tests"]
norecursedirs = [".*", "venv", "env", "*.egg", "dist", "build"]
minversion = "7.3.1"
addopts = "-rsxX -l --tb=short --strict-markers"
timeout = 300
markers = [
"init: Run the initialisation test functions",
"transfer: Run the main transfer test functions",
"exceptions: Run SQL exception test functions",
"cli: Run the cli test functions",
]
[tool.mypy]
mypy_path = "src"
python_version = "3.9"
exclude = [
"tests",
"docs",
"build",
"dist",
"venv",
"env",
]
warn_return_any = true
warn_unused_configs = true
[[tool.mypy.overrides]]
module = "pytimeparse2.*,factory.*,docker.*"
ignore_missing_imports = true07070100000020000081A4000000000000000000000001682E58C1000001D8000000000000000000000000000000000000002C00000000mysql-to-sqlite3-2.4.1/requirements_dev.txtClick>=8.1.3
docker>=6.1.3
factory-boy
Faker>=18.10.0
mysql-connector-python>=9.0.0
mysqlclient>=2.1.1
pytest>=7.3.1
pytest-cov
pytest-mock
pytest-timeout
pytimeparse2
python-dateutil>=2.9.0.post0
types_python_dateutil
python-slugify>=7.0.0
types-python-slugify
simplejson>=3.19.1
types-simplejson
sqlalchemy>=2.0.0
sqlalchemy-utils
types-sqlalchemy-utils
tox
tqdm>=4.65.0
types-tqdm
packaging
tabulate
types-tabulate
typing_extensions
requests
types-requests
mypy>=1.3.0
07070100000021000041ED000000000000000000000002682E58C100000000000000000000000000000000000000000000001B00000000mysql-to-sqlite3-2.4.1/src07070100000022000041ED000000000000000000000002682E58C100000000000000000000000000000000000000000000002C00000000mysql-to-sqlite3-2.4.1/src/mysql_to_sqlite307070100000023000081A4000000000000000000000001682E58C100000076000000000000000000000000000000000000003800000000mysql-to-sqlite3-2.4.1/src/mysql_to_sqlite3/__init__.py"""Utility to transfer data from MySQL to SQLite 3."""
__version__ = "2.4.1"
from .transporter import MySQLtoSQLite
07070100000024000081A4000000000000000000000001682E58C100001FA3000000000000000000000000000000000000003300000000mysql-to-sqlite3-2.4.1/src/mysql_to_sqlite3/cli.py"""The command line interface of MySQLtoSQLite."""
import os
import sys
import typing as t
from datetime import datetime
import click
from mysql.connector import CharacterSet
from tabulate import tabulate
from . import MySQLtoSQLite
from . import __version__ as package_version
from .click_utils import OptionEatAll, prompt_password, validate_positive_integer
from .debug_info import info
from .mysql_utils import mysql_supported_character_sets
from .sqlite_utils import CollatingSequences
_copyright_header: str = f"mysql2sqlite version {package_version} Copyright (c) 2019-{datetime.now().year} Klemen Tusar"
@click.command(
name="mysql2sqlite",
help=_copyright_header,
no_args_is_help=True,
epilog="For more information, visit https://github.com/techouse/mysql-to-sqlite3",
)
@click.option(
"-f",
"--sqlite-file",
type=click.Path(),
default=None,
help="SQLite3 database file",
required=True,
)
@click.option("-d", "--mysql-database", default=None, help="MySQL database name", required=True)
@click.option("-u", "--mysql-user", default=None, help="MySQL user", required=True)
@click.option(
"-p",
"--prompt-mysql-password",
is_flag=True,
default=False,
callback=prompt_password,
help="Prompt for MySQL password",
)
@click.option("--mysql-password", default=None, help="MySQL password")
@click.option(
"-t",
"--mysql-tables",
type=tuple,
cls=OptionEatAll,
help="Transfer only these specific tables (space separated table names). "
"Implies --without-foreign-keys which inhibits the transfer of foreign keys. "
"Can not be used together with --exclude-mysql-tables.",
)
@click.option(
"-e",
"--exclude-mysql-tables",
type=tuple,
cls=OptionEatAll,
help="Transfer all tables except these specific tables (space separated table names). "
"Implies --without-foreign-keys which inhibits the transfer of foreign keys. "
"Can not be used together with --mysql-tables.",
)
@click.option(
"-L",
"--limit-rows",
type=int,
callback=validate_positive_integer,
default=0,
help="Transfer only a limited number of rows from each table.",
)
@click.option(
"-C",
"--collation",
type=click.Choice(
[
CollatingSequences.BINARY,
CollatingSequences.NOCASE,
CollatingSequences.RTRIM,
],
case_sensitive=False,
),
default=CollatingSequences.BINARY,
show_default=True,
help="Create datatypes of TEXT affinity using a specified collation sequence.",
)
@click.option(
"-K",
"--prefix-indices",
is_flag=True,
help="Prefix indices with their corresponding tables. "
"This ensures that their names remain unique across the SQLite database.",
)
@click.option("-X", "--without-foreign-keys", is_flag=True, help="Do not transfer foreign keys.")
@click.option(
"-Z",
"--without-tables",
is_flag=True,
help="Do not transfer tables, data only.",
)
@click.option(
"-W",
"--without-data",
is_flag=True,
help="Do not transfer table data, DDL only.",
)
@click.option("-h", "--mysql-host", default="localhost", help="MySQL host. Defaults to localhost.")
@click.option("-P", "--mysql-port", type=int, default=3306, help="MySQL port. Defaults to 3306.")
@click.option(
"--mysql-charset",
metavar="TEXT",
type=click.Choice(list(CharacterSet().get_supported()), case_sensitive=False),
default="utf8mb4",
show_default=True,
help="MySQL database and table character set",
)
@click.option(
"--mysql-collation",
metavar="TEXT",
type=click.Choice(
[charset.collation for charset in mysql_supported_character_sets()],
case_sensitive=False,
),
default=None,
help="MySQL database and table collation",
)
@click.option("-S", "--skip-ssl", is_flag=True, help="Disable MySQL connection encryption.")
@click.option(
"-c",
"--chunk",
type=int,
default=200000, # this default is here for performance reasons
help="Chunk reading/writing SQL records",
)
@click.option("-l", "--log-file", type=click.Path(), help="Log file")
@click.option("--json-as-text", is_flag=True, help="Transfer JSON columns as TEXT.")
@click.option(
"-V",
"--vacuum",
is_flag=True,
help="Use the VACUUM command to rebuild the SQLite database file, "
"repacking it into a minimal amount of disk space",
)
@click.option(
"--use-buffered-cursors",
is_flag=True,
help="Use MySQLCursorBuffered for reading the MySQL database. This "
"can be useful in situations where multiple queries, with small "
"result sets, need to be combined or computed with each other.",
)
@click.option("-q", "--quiet", is_flag=True, help="Quiet. Display only errors.")
@click.option("--debug", is_flag=True, help="Debug mode. Will throw exceptions.")
@click.version_option(message=tabulate(info(), headers=["software", "version"], tablefmt="github"))
def cli(
sqlite_file: t.Union[str, "os.PathLike[t.Any]"],
mysql_user: str,
prompt_mysql_password: bool,
mysql_password: str,
mysql_database: str,
mysql_tables: t.Optional[t.Sequence[str]],
exclude_mysql_tables: t.Optional[t.Sequence[str]],
limit_rows: int,
collation: t.Optional[str],
prefix_indices: bool,
without_foreign_keys: bool,
without_tables: bool,
without_data: bool,
mysql_host: str,
mysql_port: int,
mysql_charset: str,
mysql_collation: str,
skip_ssl: bool,
chunk: int,
log_file: t.Union[str, "os.PathLike[t.Any]"],
json_as_text: bool,
vacuum: bool,
use_buffered_cursors: bool,
quiet: bool,
debug: bool,
) -> None:
"""Transfer MySQL to SQLite using the provided CLI options."""
click.echo(_copyright_header)
try:
if mysql_collation:
charset_collations: t.Tuple[str, ...] = tuple(
cs.collation for cs in mysql_supported_character_sets(mysql_charset.lower())
)
if mysql_collation not in set(charset_collations):
raise click.ClickException(
f"Error: Invalid value for '--collation' of charset '{mysql_charset}': '{mysql_collation}' "
f"""is not one of {"'" + "', '".join(charset_collations) + "'"}."""
)
# check if both mysql_skip_create_table and mysql_skip_transfer_data are True
if without_tables and without_data:
raise click.ClickException(
"Error: Both -Z/--without-tables and -W/--without-data are set. There is nothing to do. Exiting..."
)
if mysql_tables and exclude_mysql_tables:
raise click.UsageError("Illegal usage: --mysql-tables and --exclude-mysql-tables are mutually exclusive!")
converter = MySQLtoSQLite(
sqlite_file=sqlite_file,
mysql_user=mysql_user,
mysql_password=mysql_password or prompt_mysql_password,
mysql_database=mysql_database,
mysql_tables=mysql_tables,
exclude_mysql_tables=exclude_mysql_tables,
limit_rows=limit_rows,
collation=collation,
prefix_indices=prefix_indices,
without_foreign_keys=without_foreign_keys or (mysql_tables is not None and len(mysql_tables) > 0),
without_tables=without_tables,
without_data=without_data,
mysql_host=mysql_host,
mysql_port=mysql_port,
mysql_charset=mysql_charset,
mysql_collation=mysql_collation,
mysql_ssl_disabled=skip_ssl,
chunk=chunk,
json_as_text=json_as_text,
vacuum=vacuum,
buffered=use_buffered_cursors,
log_file=log_file,
quiet=quiet,
)
converter.transfer()
except KeyboardInterrupt:
if debug:
raise
click.echo("\nProcess interrupted. Exiting...")
sys.exit(1)
except Exception as err: # pylint: disable=W0703
if debug:
raise
click.echo(err)
sys.exit(1)
07070100000025000081A4000000000000000000000001682E58C100000A65000000000000000000000000000000000000003B00000000mysql-to-sqlite3-2.4.1/src/mysql_to_sqlite3/click_utils.py"""Click utilities."""
import typing as t
import click
class OptionEatAll(click.Option):
"""Taken from https://stackoverflow.com/questions/48391777/nargs-equivalent-for-options-in-click#answer-48394004.""" # noqa: ignore=E501 pylint: disable=C0301
def __init__(self, *args, **kwargs):
"""Override."""
self.save_other_options = kwargs.pop("save_other_options", True)
nargs = kwargs.pop("nargs", -1)
if nargs != -1:
raise ValueError(f"nargs, if set, must be -1 not {nargs}")
super(OptionEatAll, self).__init__(*args, **kwargs)
self._previous_parser_process = None
self._eat_all_parser = None
def add_to_parser(self, parser, ctx) -> None:
"""Override."""
def parser_process(value, state):
# method to hook to the parser.process
done = False
value = [value]
if self.save_other_options:
# grab everything up to the next option
while state.rargs and not done:
for prefix in self._eat_all_parser.prefixes:
if state.rargs[0].startswith(prefix):
done = True
if not done:
value.append(state.rargs.pop(0))
else:
# grab everything remaining
value += state.rargs
state.rargs[:] = []
value = tuple(value)
# call the actual process
self._previous_parser_process(value, state)
retval = super(OptionEatAll, self).add_to_parser(parser, ctx) # pylint: disable=E1111
for name in self.opts:
# pylint: disable=W0212
our_parser = parser._long_opt.get(name) or parser._short_opt.get(name)
if our_parser:
self._eat_all_parser = our_parser
self._previous_parser_process = our_parser.process
our_parser.process = parser_process
break
return retval
def prompt_password(ctx: click.core.Context, param: t.Any, use_password: bool): # pylint: disable=W0613
"""Prompt for password."""
if use_password:
mysql_password = ctx.params.get("mysql_password")
if not mysql_password:
mysql_password = click.prompt("MySQL password", hide_input=True)
return mysql_password
def validate_positive_integer(ctx: click.core.Context, param: t.Any, value: int): # pylint: disable=W0613
"""Allow only positive integers and 0."""
if value < 0:
raise click.BadParameter("Should be a positive integer or 0.")
return value
07070100000026000081A4000000000000000000000001682E58C100000D07000000000000000000000000000000000000003A00000000mysql-to-sqlite3-2.4.1/src/mysql_to_sqlite3/debug_info.py"""Module containing bug report helper(s).
Adapted from https://github.com/psf/requests/blob/master/requests/help.py
"""
import platform
import sqlite3
import sys
import typing as t
from shutil import which
from subprocess import check_output
import click
import mysql.connector
import pytimeparse2
import simplejson
import slugify
import tabulate
import tqdm
from . import __version__ as package_version
def _implementation() -> str:
"""Return a dict with the Python implementation and version.
Provide both the name and the version of the Python implementation
currently running. For example, on CPython 2.7.5 it will return
{'name': 'CPython', 'version': '2.7.5'}.
This function works best on CPython and PyPy: in particular, it probably
doesn't work for Jython or IronPython. Future investigation should be done
to work out the correct shape of the code for those platforms.
"""
implementation: str = platform.python_implementation()
if implementation == "CPython":
implementation_version = platform.python_version()
elif implementation == "PyPy":
implementation_version = "%s.%s.%s" % (
sys.pypy_version_info.major, # type: ignore # noqa: ignore=E1101 pylint: disable=E1101
sys.pypy_version_info.minor, # type: ignore # noqa: ignore=E1101 pylint: disable=E1101
sys.pypy_version_info.micro, # type: ignore # noqa: ignore=E1101 pylint: disable=E1101
)
rel = sys.pypy_version_info.releaselevel # type: ignore # noqa: ignore=E1101 pylint: disable=E1101
if rel != "final":
implementation_version = "".join([implementation_version, rel])
elif implementation == "Jython":
implementation_version = platform.python_version() # Complete Guess
elif implementation == "IronPython":
implementation_version = platform.python_version() # Complete Guess
else:
implementation_version = "Unknown"
return f"{implementation} {implementation_version}"
def _mysql_version() -> str:
if which("mysql") is not None:
try:
mysql_version: t.Union[str, bytes] = check_output(["mysql", "-V"])
try:
return mysql_version.decode().strip() # type: ignore
except (UnicodeDecodeError, AttributeError):
return str(mysql_version)
except Exception: # nosec pylint: disable=W0703
pass
return "MySQL client not found on the system"
def info() -> t.List[t.List[str]]:
"""Generate information for a bug report."""
try:
platform_info: str = f"{platform.system()} {platform.release()}"
except IOError:
platform_info = "Unknown"
return [
["mysql-to-sqlite3", package_version],
["", ""],
["Operating System", platform_info],
["Python", _implementation()],
["MySQL", _mysql_version()],
["SQLite", sqlite3.sqlite_version],
["", ""],
["click", str(click.__version__)],
["mysql-connector-python", mysql.connector.__version__],
["python-slugify", slugify.__version__],
["pytimeparse2", pytimeparse2.__version__],
["simplejson", simplejson.__version__], # type: ignore
["tabulate", tabulate.__version__],
["tqdm", tqdm.__version__],
]
07070100000027000081A4000000000000000000000001682E58C1000004E4000000000000000000000000000000000000003B00000000mysql-to-sqlite3-2.4.1/src/mysql_to_sqlite3/mysql_utils.py"""Miscellaneous MySQL utilities."""
import typing as t
from mysql.connector import CharacterSet
from mysql.connector.charsets import MYSQL_CHARACTER_SETS
CHARSET_INTRODUCERS: t.Tuple[str, ...] = tuple(
f"_{charset[0]}" for charset in MYSQL_CHARACTER_SETS if charset is not None
)
class CharSet(t.NamedTuple):
"""MySQL character set as a named tuple."""
id: int
charset: str
collation: str
def mysql_supported_character_sets(charset: t.Optional[str] = None) -> t.Iterator[CharSet]:
"""Get supported MySQL character sets."""
index: int
info: t.Optional[t.Tuple[str, str, bool]]
if charset is not None:
for index, info in enumerate(MYSQL_CHARACTER_SETS):
if info is not None:
try:
if info[0] == charset:
yield CharSet(index, charset, info[1])
except KeyError:
continue
else:
for charset in CharacterSet().get_supported():
for index, info in enumerate(MYSQL_CHARACTER_SETS):
if info is not None:
try:
yield CharSet(index, charset, info[1])
except KeyError:
continue
07070100000028000081A4000000000000000000000001682E58C100000000000000000000000000000000000000000000003500000000mysql-to-sqlite3-2.4.1/src/mysql_to_sqlite3/py.typed07070100000029000081A4000000000000000000000001682E58C10000086A000000000000000000000000000000000000003C00000000mysql-to-sqlite3-2.4.1/src/mysql_to_sqlite3/sqlite_utils.py"""SQLite adapters and converters for unsupported data types."""
import sqlite3
import typing as t
from datetime import date, timedelta
from decimal import Decimal
from dateutil.parser import ParserError
from dateutil.parser import parse as dateutil_parse
from pytimeparse2 import parse
def adapt_decimal(value: t.Any) -> str:
"""Convert decimal.Decimal to string."""
return str(value)
def convert_decimal(value: t.Any) -> Decimal:
"""Convert string to decimal.Decimal."""
return Decimal(value)
def adapt_timedelta(value: t.Any) -> str:
"""Convert datetime.timedelta to %H:%M:%S string."""
hours, remainder = divmod(value.total_seconds(), 3600)
minutes, seconds = divmod(remainder, 60)
return "{:02}:{:02}:{:02}".format(int(hours), int(minutes), int(seconds))
def convert_timedelta(value: t.Any) -> timedelta:
"""Convert %H:%M:%S string to datetime.timedelta."""
return timedelta(seconds=parse(value))
def encode_data_for_sqlite(value: t.Any) -> t.Any:
"""Fix encoding bytes."""
if isinstance(value, bytes):
try:
return value.decode()
except (UnicodeDecodeError, AttributeError):
return sqlite3.Binary(value)
elif isinstance(value, str):
return value
else:
try:
return sqlite3.Binary(value)
except TypeError:
return value
class CollatingSequences:
"""Taken from https://www.sqlite.org/datatype3.html#collating_sequences."""
BINARY: str = "BINARY"
NOCASE: str = "NOCASE"
RTRIM: str = "RTRIM"
def convert_date(value: t.Union[str, bytes]) -> date:
"""Handle SQLite date conversion."""
try:
return dateutil_parse(value.decode() if isinstance(value, bytes) else value).date()
except ParserError as err:
raise ValueError(f"DATE field contains {err}") # pylint: disable=W0707
Integer_Types: t.Set[str] = {
"INTEGER",
"INTEGER UNSIGNED",
"INT",
"INT UNSIGNED",
"BIGINT",
"BIGINT UNSIGNED",
"MEDIUMINT",
"MEDIUMINT UNSIGNED",
"SMALLINT",
"SMALLINT UNSIGNED",
"TINYINT",
"TINYINT UNSIGNED",
"NUMERIC",
}
0707010000002A000081A4000000000000000000000001682E58C100007EFD000000000000000000000000000000000000003B00000000mysql-to-sqlite3-2.4.1/src/mysql_to_sqlite3/transporter.py"""Use to transfer a MySQL database to SQLite."""
import logging
import os
import re
import sqlite3
import typing as t
from datetime import timedelta
from decimal import Decimal
from math import ceil
from os.path import realpath
from sys import stdout
import mysql.connector
import typing_extensions as tx
from mysql.connector import CharacterSet, errorcode
from mysql.connector.abstracts import MySQLConnectionAbstract
from mysql.connector.types import RowItemType
from tqdm import tqdm, trange
from mysql_to_sqlite3.mysql_utils import CHARSET_INTRODUCERS
from mysql_to_sqlite3.sqlite_utils import (
CollatingSequences,
Integer_Types,
adapt_decimal,
adapt_timedelta,
convert_date,
convert_decimal,
convert_timedelta,
encode_data_for_sqlite,
)
from mysql_to_sqlite3.types import MySQLtoSQLiteAttributes, MySQLtoSQLiteParams
class MySQLtoSQLite(MySQLtoSQLiteAttributes):
"""Use this class to transfer a MySQL database to SQLite."""
COLUMN_PATTERN: t.Pattern[str] = re.compile(r"^[^(]+")
COLUMN_LENGTH_PATTERN: t.Pattern[str] = re.compile(r"\(\d+\)$")
def __init__(self, **kwargs: tx.Unpack[MySQLtoSQLiteParams]) -> None:
"""Constructor."""
if kwargs.get("mysql_database") is not None:
self._mysql_database = str(kwargs.get("mysql_database"))
else:
raise ValueError("Please provide a MySQL database")
if kwargs.get("mysql_user") is not None:
self._mysql_user = str(kwargs.get("mysql_user"))
else:
raise ValueError("Please provide a MySQL user")
if kwargs.get("sqlite_file") is None:
raise ValueError("Please provide an SQLite file")
else:
self._sqlite_file = realpath(str(kwargs.get("sqlite_file")))
password: t.Optional[t.Union[str, bool]] = kwargs.get("mysql_password")
self._mysql_password = password if isinstance(password, str) else None
self._mysql_host = kwargs.get("mysql_host", "localhost") or "localhost"
self._mysql_port = kwargs.get("mysql_port", 3306) or 3306
self._mysql_charset = kwargs.get("mysql_charset", "utf8mb4") or "utf8mb4"
self._mysql_collation = (
kwargs.get("mysql_collation") or CharacterSet().get_default_collation(self._mysql_charset.lower())[0]
)
if not kwargs.get("mysql_collation") and self._mysql_collation == "utf8mb4_0900_ai_ci":
self._mysql_collation = "utf8mb4_unicode_ci"
self._mysql_tables = kwargs.get("mysql_tables") or tuple()
self._exclude_mysql_tables = kwargs.get("exclude_mysql_tables") or tuple()
if bool(self._mysql_tables) and bool(self._exclude_mysql_tables):
raise ValueError("mysql_tables and exclude_mysql_tables are mutually exclusive")
self._limit_rows = kwargs.get("limit_rows", 0) or 0
if kwargs.get("collation") is not None and str(kwargs.get("collation")).upper() in {
CollatingSequences.BINARY,
CollatingSequences.NOCASE,
CollatingSequences.RTRIM,
}:
self._collation = str(kwargs.get("collation")).upper()
else:
self._collation = CollatingSequences.BINARY
self._prefix_indices = kwargs.get("prefix_indices", False) or False
if bool(self._mysql_tables) or bool(self._exclude_mysql_tables):
self._without_foreign_keys = True
else:
self._without_foreign_keys = bool(kwargs.get("without_foreign_keys", False))
self._without_data = bool(kwargs.get("without_data", False))
self._without_tables = bool(kwargs.get("without_tables", False))
if self._without_tables and self._without_data:
raise ValueError("Unable to continue without transferring data or creating tables!")
self._mysql_ssl_disabled = bool(kwargs.get("mysql_ssl_disabled", False))
self._current_chunk_number = 0
self._chunk_size = kwargs.get("chunk") or None
self._buffered = bool(kwargs.get("buffered", False))
self._vacuum = bool(kwargs.get("vacuum", False))
self._quiet = bool(kwargs.get("quiet", False))
self._logger = self._setup_logger(log_file=kwargs.get("log_file") or None, quiet=self._quiet)
sqlite3.register_adapter(Decimal, adapt_decimal)
sqlite3.register_converter("DECIMAL", convert_decimal)
sqlite3.register_adapter(timedelta, adapt_timedelta)
sqlite3.register_converter("DATE", convert_date)
sqlite3.register_converter("TIME", convert_timedelta)
self._sqlite = sqlite3.connect(realpath(self._sqlite_file), detect_types=sqlite3.PARSE_DECLTYPES)
self._sqlite.row_factory = sqlite3.Row
self._sqlite_cur = self._sqlite.cursor()
self._json_as_text = bool(kwargs.get("json_as_text", False))
self._sqlite_json1_extension_enabled = not self._json_as_text and self._check_sqlite_json1_extension_enabled()
try:
_mysql_connection = mysql.connector.connect(
user=self._mysql_user,
password=self._mysql_password,
host=self._mysql_host,
port=self._mysql_port,
ssl_disabled=self._mysql_ssl_disabled,
charset=self._mysql_charset,
collation=self._mysql_collation,
)
if isinstance(_mysql_connection, MySQLConnectionAbstract):
self._mysql = _mysql_connection
else:
raise ConnectionError("Unable to connect to MySQL")
if not self._mysql.is_connected():
raise ConnectionError("Unable to connect to MySQL")
self._mysql_cur = self._mysql.cursor(buffered=self._buffered, raw=True) # type: ignore[assignment]
self._mysql_cur_prepared = self._mysql.cursor(prepared=True) # type: ignore[assignment]
self._mysql_cur_dict = self._mysql.cursor( # type: ignore[assignment]
buffered=self._buffered,
dictionary=True,
)
try:
self._mysql.database = self._mysql_database
except (mysql.connector.Error, Exception) as err:
if hasattr(err, "errno") and err.errno == errorcode.ER_BAD_DB_ERROR:
self._logger.error("MySQL Database does not exist!")
raise
self._logger.error(err)
raise
except mysql.connector.Error as err:
self._logger.error(err)
raise
@classmethod
def _setup_logger(
cls, log_file: t.Optional[t.Union[str, "os.PathLike[t.Any]"]] = None, quiet: bool = False
) -> logging.Logger:
formatter: logging.Formatter = logging.Formatter(
fmt="%(asctime)s %(levelname)-8s %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
)
logger: logging.Logger = logging.getLogger(cls.__name__)
logger.setLevel(logging.DEBUG)
if not quiet:
screen_handler = logging.StreamHandler(stream=stdout)
screen_handler.setFormatter(formatter)
logger.addHandler(screen_handler)
if log_file:
file_handler = logging.FileHandler(realpath(log_file), mode="w")
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
return logger
@classmethod
def _valid_column_type(cls, column_type: str) -> t.Optional[t.Match[str]]:
return cls.COLUMN_PATTERN.match(column_type.strip())
@classmethod
def _column_type_length(cls, column_type: str) -> str:
suffix: t.Optional[t.Match[str]] = cls.COLUMN_LENGTH_PATTERN.search(column_type)
if suffix:
return suffix.group(0)
return ""
@staticmethod
def _decode_column_type(column_type: t.Union[str, bytes]) -> str:
if isinstance(column_type, str):
return column_type
if isinstance(column_type, bytes):
try:
return column_type.decode()
except (UnicodeDecodeError, AttributeError):
pass
return str(column_type)
@classmethod
def _translate_type_from_mysql_to_sqlite(
cls, column_type: t.Union[str, bytes], sqlite_json1_extension_enabled=False
) -> str:
_column_type: str = cls._decode_column_type(column_type)
# This could be optimized even further, however is seems adequate.
match: t.Optional[t.Match[str]] = cls._valid_column_type(_column_type)
if not match:
raise ValueError(f'"{_column_type}" is not a valid column_type!')
data_type: str = match.group(0).upper()
if data_type.endswith(" UNSIGNED"):
data_type = data_type.replace(" UNSIGNED", "")
if data_type in {
"BIGINT",
"BLOB",
"BOOLEAN",
"DATE",
"DATETIME",
"DECIMAL",
"DOUBLE",
"FLOAT",
"INTEGER",
"MEDIUMINT",
"NUMERIC",
"REAL",
"SMALLINT",
"TIME",
"TINYINT",
"YEAR",
}:
return data_type
if data_type in {
"BIT",
"BINARY",
"LONGBLOB",
"MEDIUMBLOB",
"TINYBLOB",
"VARBINARY",
}:
return "BLOB"
if data_type in {"NCHAR", "NVARCHAR", "VARCHAR"}:
return data_type + cls._column_type_length(_column_type)
if data_type == "CHAR":
return "CHARACTER" + cls._column_type_length(_column_type)
if data_type == "INT":
return "INTEGER"
if data_type in "TIMESTAMP":
return "DATETIME"
if data_type == "JSON" and sqlite_json1_extension_enabled:
return "JSON"
return "TEXT"
@classmethod
def _translate_default_from_mysql_to_sqlite(
cls,
column_default: RowItemType = None,
column_type: t.Optional[str] = None,
column_extra: RowItemType = None,
) -> str:
is_binary: bool
is_hex: bool
if isinstance(column_default, bytes):
if column_type in {
"BIT",
"BINARY",
"BLOB",
"LONGBLOB",
"MEDIUMBLOB",
"TINYBLOB",
"VARBINARY",
}:
if column_extra in {"DEFAULT_GENERATED", "default_generated"}:
for charset_introducer in CHARSET_INTRODUCERS:
if column_default.startswith(charset_introducer.encode()):
is_binary = False
is_hex = False
for b_prefix in ("B", "b"):
if column_default.startswith(rf"{charset_introducer} {b_prefix}\'".encode()):
is_binary = True
break
for x_prefix in ("X", "x"):
if column_default.startswith(rf"{charset_introducer} {x_prefix}\'".encode()):
is_hex = True
break
column_default = (
column_default.replace(charset_introducer.encode(), b"")
.replace(rb"x\'", b"")
.replace(rb"X\'", b"")
.replace(rb"b\'", b"")
.replace(rb"B\'", b"")
.replace(rb"\'", b"")
.replace(rb"'", b"")
.strip()
)
if is_binary:
return f"DEFAULT '{chr(int(column_default, 2))}'"
if is_hex:
return f"DEFAULT x'{column_default.decode()}'"
break
return f"DEFAULT x'{column_default.hex()}'"
try:
column_default = column_default.decode()
except (UnicodeDecodeError, AttributeError):
pass
if column_default is None:
return ""
if isinstance(column_default, bool):
if column_type == "BOOLEAN" and sqlite3.sqlite_version >= "3.23.0":
if column_default:
return "DEFAULT(TRUE)"
return "DEFAULT(FALSE)"
return f"DEFAULT '{int(column_default)}'"
if isinstance(column_default, str):
if column_extra in {"DEFAULT_GENERATED", "default_generated"}:
if column_default.upper() in {
"CURRENT_TIME",
"CURRENT_DATE",
"CURRENT_TIMESTAMP",
}:
return f"DEFAULT {column_default.upper()}"
for charset_introducer in CHARSET_INTRODUCERS:
if column_default.startswith(charset_introducer):
is_binary = False
is_hex = False
for b_prefix in ("B", "b"):
if column_default.startswith(rf"{charset_introducer} {b_prefix}\'"):
is_binary = True
break
for x_prefix in ("X", "x"):
if column_default.startswith(rf"{charset_introducer} {x_prefix}\'"):
is_hex = True
break
column_default = (
column_default.replace(charset_introducer, "")
.replace(r"x\'", "")
.replace(r"X\'", "")
.replace(r"b\'", "")
.replace(r"B\'", "")
.replace(r"\'", "")
.replace(r"'", "")
.strip()
)
if is_binary:
return f"DEFAULT '{chr(int(column_default, 2))}'"
if is_hex:
return f"DEFAULT x'{column_default}'"
return f"DEFAULT '{column_default}'"
return "DEFAULT '{}'".format(column_default.replace(r"\'", r"''"))
return "DEFAULT '{}'".format(str(column_default).replace(r"\'", r"''"))
@classmethod
def _data_type_collation_sequence(
cls, collation: str = CollatingSequences.BINARY, column_type: t.Optional[str] = None
) -> str:
if column_type and collation != CollatingSequences.BINARY:
if column_type.startswith(
(
"CHARACTER",
"NCHAR",
"NVARCHAR",
"TEXT",
"VARCHAR",
)
):
return f"COLLATE {collation}"
return ""
def _check_sqlite_json1_extension_enabled(self) -> bool:
try:
self._sqlite_cur.execute("PRAGMA compile_options")
return "ENABLE_JSON1" in set(row[0] for row in self._sqlite_cur.fetchall())
except sqlite3.Error:
return False
def _build_create_table_sql(self, table_name: str) -> str:
sql: str = f'CREATE TABLE IF NOT EXISTS "{table_name}" ('
primary: str = ""
indices: str = ""
self._mysql_cur_dict.execute(f"SHOW COLUMNS FROM `{table_name}`")
rows: t.Sequence[t.Optional[t.Dict[str, RowItemType]]] = self._mysql_cur_dict.fetchall()
primary_keys: int = sum(1 for row in rows if row is not None and row["Key"] == "PRI")
for row in rows:
if row is not None:
column_type = self._translate_type_from_mysql_to_sqlite(
column_type=row["Type"], # type: ignore[arg-type]
sqlite_json1_extension_enabled=self._sqlite_json1_extension_enabled,
)
if row["Key"] == "PRI" and row["Extra"] == "auto_increment" and primary_keys == 1:
if column_type in Integer_Types:
sql += '\n\t"{name}" INTEGER PRIMARY KEY AUTOINCREMENT,'.format(
name=row["Field"].decode() if isinstance(row["Field"], bytes) else row["Field"],
)
else:
self._logger.warning(
'Primary key "%s" in table "%s" is not an INTEGER type! Skipping.',
row["Field"],
table_name,
)
else:
sql += '\n\t"{name}" {type} {notnull} {default} {collation},'.format(
name=row["Field"].decode() if isinstance(row["Field"], bytes) else row["Field"],
type=column_type,
notnull="NULL" if row["Null"] == "YES" else "NOT NULL",
default=self._translate_default_from_mysql_to_sqlite(row["Default"], column_type, row["Extra"]),
collation=self._data_type_collation_sequence(self._collation, column_type),
)
self._mysql_cur_dict.execute(
"""
SELECT s.INDEX_NAME AS `name`,
IF (NON_UNIQUE = 0 AND s.INDEX_NAME = 'PRIMARY', 1, 0) AS `primary`,
IF (NON_UNIQUE = 0 AND s.INDEX_NAME <> 'PRIMARY', 1, 0) AS `unique`,
{auto_increment}
GROUP_CONCAT(s.COLUMN_NAME ORDER BY SEQ_IN_INDEX) AS `columns`,
GROUP_CONCAT(c.COLUMN_TYPE ORDER BY SEQ_IN_INDEX) AS `types`
FROM information_schema.STATISTICS AS s
JOIN information_schema.COLUMNS AS c
ON s.TABLE_SCHEMA = c.TABLE_SCHEMA
AND s.TABLE_NAME = c.TABLE_NAME
AND s.COLUMN_NAME = c.COLUMN_NAME
WHERE s.TABLE_SCHEMA = %s
AND s.TABLE_NAME = %s
GROUP BY s.INDEX_NAME, s.NON_UNIQUE {group_by_extra}
""".format(
auto_increment=(
"IF (c.EXTRA = 'auto_increment', 1, 0) AS `auto_increment`,"
if primary_keys == 1
else "0 as `auto_increment`,"
),
group_by_extra=" ,c.EXTRA" if primary_keys == 1 else "",
),
(self._mysql_database, table_name),
)
mysql_indices: t.Sequence[t.Optional[t.Dict[str, RowItemType]]] = self._mysql_cur_dict.fetchall()
for index in mysql_indices:
if index is not None:
index_name: str
if isinstance(index["name"], bytes):
index_name = index["name"].decode()
elif isinstance(index["name"], str):
index_name = index["name"]
else:
index_name = str(index["name"])
# check if the index name collides with any table name
self._mysql_cur_dict.execute(
"""
SELECT COUNT(*) AS `count`
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = %s
AND TABLE_NAME = %s
""",
(self._mysql_database, index_name),
)
collision: t.Optional[t.Dict[str, RowItemType]] = self._mysql_cur_dict.fetchone()
table_collisions: int = 0
if collision is not None:
table_collisions = int(collision["count"]) # type: ignore[arg-type]
columns: str = ""
if isinstance(index["columns"], bytes):
columns = index["columns"].decode()
elif isinstance(index["columns"], str):
columns = index["columns"]
types: str = ""
if isinstance(index["types"], bytes):
types = index["types"].decode()
elif isinstance(index["types"], str):
types = index["types"]
if len(columns) > 0:
if index["primary"] in {1, "1"}:
if (index["auto_increment"] not in {1, "1"}) or any(
self._translate_type_from_mysql_to_sqlite(
column_type=_type,
sqlite_json1_extension_enabled=self._sqlite_json1_extension_enabled,
)
not in Integer_Types
for _type in types.split(",")
):
primary += "\n\tPRIMARY KEY ({columns})".format(
columns=", ".join(f'"{column}"' for column in columns.split(","))
)
else:
indices += """CREATE {unique} INDEX IF NOT EXISTS "{name}" ON "{table}" ({columns});""".format(
unique="UNIQUE" if index["unique"] in {1, "1"} else "",
name=(
f"{table_name}_{index_name}"
if (table_collisions > 0 or self._prefix_indices)
else index_name
),
table=table_name,
columns=", ".join(f'"{column}"' for column in columns.split(",")),
)
sql += primary
sql = sql.rstrip(", ")
if not self._without_tables and not self._without_foreign_keys:
server_version: t.Optional[t.Tuple[int, ...]] = self._mysql.get_server_version()
self._mysql_cur_dict.execute(
"""
SELECT k.COLUMN_NAME AS `column`,
k.REFERENCED_TABLE_NAME AS `ref_table`,
k.REFERENCED_COLUMN_NAME AS `ref_column`,
c.UPDATE_RULE AS `on_update`,
c.DELETE_RULE AS `on_delete`
FROM information_schema.TABLE_CONSTRAINTS AS i
{JOIN} information_schema.KEY_COLUMN_USAGE AS k
ON i.CONSTRAINT_NAME = k.CONSTRAINT_NAME
AND i.TABLE_NAME = k.TABLE_NAME
{JOIN} information_schema.REFERENTIAL_CONSTRAINTS AS c
ON c.CONSTRAINT_NAME = i.CONSTRAINT_NAME
AND c.TABLE_NAME = i.TABLE_NAME
WHERE i.TABLE_SCHEMA = %s
AND i.TABLE_NAME = %s
AND i.CONSTRAINT_TYPE = %s
GROUP BY i.CONSTRAINT_NAME,
k.COLUMN_NAME,
k.REFERENCED_TABLE_NAME,
k.REFERENCED_COLUMN_NAME,
c.UPDATE_RULE,
c.DELETE_RULE
""".format(
JOIN=(
"JOIN"
if (server_version is not None and server_version[0] == 8 and server_version[2] > 19)
else "LEFT JOIN"
)
),
(self._mysql_database, table_name, "FOREIGN KEY"),
)
for foreign_key in self._mysql_cur_dict.fetchall():
if foreign_key is not None:
sql += (
',\n\tFOREIGN KEY("{column}") REFERENCES "{ref_table}" ("{ref_column}") '
"ON UPDATE {on_update} "
"ON DELETE {on_delete}".format(**foreign_key) # type: ignore[str-bytes-safe]
)
sql += "\n);"
sql += indices
return sql
def _create_table(self, table_name: str, attempting_reconnect: bool = False) -> None:
try:
if attempting_reconnect:
self._mysql.reconnect()
self._sqlite_cur.executescript(self._build_create_table_sql(table_name))
self._sqlite.commit()
except mysql.connector.Error as err:
if err.errno == errorcode.CR_SERVER_LOST:
if not attempting_reconnect:
self._logger.warning("Connection to MySQL server lost.\nAttempting to reconnect.")
self._create_table(table_name, True)
else:
self._logger.warning("Connection to MySQL server lost.\nReconnection attempt aborted.")
raise
self._logger.error(
"MySQL failed reading table definition from table %s: %s",
table_name,
err,
)
raise
except sqlite3.Error as err:
self._logger.error("SQLite failed creating table %s: %s", table_name, err)
raise
def _transfer_table_data(
self, table_name: str, sql: str, total_records: int = 0, attempting_reconnect: bool = False
) -> None:
if attempting_reconnect:
self._mysql.reconnect()
try:
if self._chunk_size is not None and self._chunk_size > 0:
for chunk in trange(
self._current_chunk_number,
int(ceil(total_records / self._chunk_size)),
disable=self._quiet,
):
self._current_chunk_number = chunk
self._sqlite_cur.executemany(
sql,
(
tuple(encode_data_for_sqlite(col) if col is not None else None for col in row)
for row in self._mysql_cur.fetchmany(self._chunk_size)
),
)
else:
self._sqlite_cur.executemany(
sql,
(
tuple(encode_data_for_sqlite(col) if col is not None else None for col in row)
for row in tqdm(
self._mysql_cur.fetchall(),
total=total_records,
disable=self._quiet,
)
),
)
self._sqlite.commit()
except mysql.connector.Error as err:
if err.errno == errorcode.CR_SERVER_LOST:
if not attempting_reconnect:
self._logger.warning("Connection to MySQL server lost.\nAttempting to reconnect.")
self._transfer_table_data(
table_name=table_name,
sql=sql,
total_records=total_records,
attempting_reconnect=True,
)
else:
self._logger.warning("Connection to MySQL server lost.\nReconnection attempt aborted.")
raise
self._logger.error(
"MySQL transfer failed reading table data from table %s: %s",
table_name,
err,
)
raise
except sqlite3.Error as err:
self._logger.error(
"SQLite transfer failed inserting data into table %s: %s",
table_name,
err,
)
raise
def transfer(self) -> None:
"""The primary and only method with which we transfer all the data."""
if len(self._mysql_tables) > 0 or len(self._exclude_mysql_tables) > 0:
# transfer only specific tables
specific_tables: t.Sequence[str] = (
self._exclude_mysql_tables if len(self._exclude_mysql_tables) > 0 else self._mysql_tables
)
self._mysql_cur_prepared.execute(
"""
SELECT TABLE_NAME
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = SCHEMA()
AND TABLE_NAME {exclude} IN ({placeholders})
""".format(
exclude="NOT" if len(self._exclude_mysql_tables) > 0 else "",
placeholders=("%s, " * len(specific_tables)).rstrip(" ,"),
),
specific_tables,
)
tables: t.Iterable[RowItemType] = (row[0] for row in self._mysql_cur_prepared.fetchall())
else:
# transfer all tables
self._mysql_cur.execute(
"""
SELECT TABLE_NAME
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = SCHEMA()
"""
)
tables = (row[0].decode() for row in self._mysql_cur.fetchall()) # type: ignore[union-attr]
try:
# turn off foreign key checking in SQLite while transferring data
self._sqlite_cur.execute("PRAGMA foreign_keys=OFF")
for table_name in tables:
if isinstance(table_name, bytes):
table_name = table_name.decode()
self._logger.info(
"%s%sTransferring table %s",
"[WITHOUT DATA] " if self._without_data else "",
"[ONLY DATA] " if self._without_tables else "",
table_name,
)
# reset the chunk
self._current_chunk_number = 0
if not self._without_tables:
# create the table
self._create_table(table_name) # type: ignore[arg-type]
if not self._without_data:
# get the size of the data
if self._limit_rows > 0:
# limit to the requested number of rows
self._mysql_cur_dict.execute(
"SELECT COUNT(*) AS `total_records` "
f"FROM (SELECT * FROM `{table_name}` LIMIT {self._limit_rows}) AS `table`"
)
else:
# get all rows
self._mysql_cur_dict.execute(f"SELECT COUNT(*) AS `total_records` FROM `{table_name}`")
total_records: t.Optional[t.Dict[str, RowItemType]] = self._mysql_cur_dict.fetchone()
if total_records is not None:
total_records_count: int = int(total_records["total_records"]) # type: ignore[arg-type]
else:
total_records_count = 0
# only continue if there is anything to transfer
if total_records_count > 0:
# populate it
self._mysql_cur.execute(
"SELECT * FROM `{table_name}` {limit}".format(
table_name=table_name,
limit=f"LIMIT {self._limit_rows}" if self._limit_rows > 0 else "",
)
)
columns: t.Tuple[str, ...] = tuple(column[0] for column in self._mysql_cur.description) # type: ignore[union-attr]
# build the SQL string
sql = """
INSERT OR IGNORE
INTO "{table}" ({fields})
VALUES ({placeholders})
""".format(
table=table_name,
fields=('"{}", ' * len(columns)).rstrip(" ,").format(*columns),
placeholders=("?, " * len(columns)).rstrip(" ,"),
)
self._transfer_table_data(
table_name=table_name, # type: ignore[arg-type]
sql=sql,
total_records=total_records_count,
)
except Exception: # pylint: disable=W0706
raise
finally:
# re-enable foreign key checking once done transferring
self._sqlite_cur.execute("PRAGMA foreign_keys=ON")
if self._vacuum:
self._logger.info("Vacuuming created SQLite database file.\nThis might take a while.")
self._sqlite_cur.execute("VACUUM")
self._logger.info("Done!")
0707010000002B000081A4000000000000000000000001682E58C10000087F000000000000000000000000000000000000003500000000mysql-to-sqlite3-2.4.1/src/mysql_to_sqlite3/types.py"""Types for mysql-to-sqlite3."""
import os
import typing as t
from logging import Logger
from sqlite3 import Connection, Cursor
import typing_extensions as tx
from mysql.connector.abstracts import MySQLConnectionAbstract
from mysql.connector.cursor import MySQLCursorDict, MySQLCursorPrepared, MySQLCursorRaw
class MySQLtoSQLiteParams(tx.TypedDict):
"""MySQLtoSQLite parameters."""
buffered: t.Optional[bool]
chunk: t.Optional[int]
collation: t.Optional[str]
exclude_mysql_tables: t.Optional[t.Sequence[str]]
json_as_text: t.Optional[bool]
limit_rows: t.Optional[int]
log_file: t.Optional[t.Union[str, "os.PathLike[t.Any]"]]
mysql_database: str
mysql_host: str
mysql_password: t.Optional[t.Union[str, bool]]
mysql_port: int
mysql_charset: t.Optional[str]
mysql_collation: t.Optional[str]
mysql_ssl_disabled: t.Optional[bool]
mysql_tables: t.Optional[t.Sequence[str]]
mysql_user: str
prefix_indices: t.Optional[bool]
quiet: t.Optional[bool]
sqlite_file: t.Union[str, "os.PathLike[t.Any]"]
vacuum: t.Optional[bool]
without_tables: t.Optional[bool]
without_data: t.Optional[bool]
without_foreign_keys: t.Optional[bool]
class MySQLtoSQLiteAttributes:
"""MySQLtoSQLite attributes."""
_buffered: bool
_chunk_size: t.Optional[int]
_collation: str
_current_chunk_number: int
_exclude_mysql_tables: t.Sequence[str]
_json_as_text: bool
_limit_rows: int
_logger: Logger
_mysql: MySQLConnectionAbstract
_mysql_cur: MySQLCursorRaw
_mysql_cur_dict: MySQLCursorDict
_mysql_cur_prepared: MySQLCursorPrepared
_mysql_database: str
_mysql_host: str
_mysql_password: t.Optional[str]
_mysql_port: int
_mysql_charset: str
_mysql_collation: str
_mysql_ssl_disabled: bool
_mysql_tables: t.Sequence[str]
_mysql_user: str
_prefix_indices: bool
_quiet: bool
_sqlite: Connection
_sqlite_cur: Cursor
_sqlite_file: t.Union[str, "os.PathLike[t.Any]"]
_without_tables: bool
_sqlite_json1_extension_enabled: bool
_vacuum: bool
_without_data: bool
_without_foreign_keys: bool
0707010000002C000041ED000000000000000000000002682E58C100000000000000000000000000000000000000000000001D00000000mysql-to-sqlite3-2.4.1/tests0707010000002D000081A4000000000000000000000001682E58C100000000000000000000000000000000000000000000002900000000mysql-to-sqlite3-2.4.1/tests/__init__.py0707010000002E000081A4000000000000000000000001682E58C100002ABA000000000000000000000000000000000000002900000000mysql-to-sqlite3-2.4.1/tests/conftest.pyimport json
import os
import socket
import typing as t
from codecs import open
from contextlib import contextmanager
from os.path import abspath, dirname, isfile, join
from pathlib import Path
from random import choice
from string import ascii_lowercase, ascii_uppercase, digits
from time import sleep
import docker
import mysql.connector
import pytest
from _pytest._py.path import LocalPath
from _pytest.config import Config
from _pytest.config.argparsing import Parser
from _pytest.legacypath import TempdirFactory
from click.testing import CliRunner
from docker import DockerClient
from docker.errors import NotFound
from docker.models.containers import Container
from faker import Faker
from mysql.connector import MySQLConnection, errorcode
from mysql.connector.connection_cext import CMySQLConnection
from mysql.connector.pooling import PooledMySQLConnection
from requests import HTTPError
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from sqlalchemy_utils import database_exists, drop_database
from . import database, factories, models
def pytest_addoption(parser: "Parser"):
parser.addoption(
"--mysql-user",
dest="mysql_user",
default="tester",
help="MySQL user. Defaults to 'tester'.",
)
parser.addoption(
"--mysql-password",
dest="mysql_password",
default="testpass",
help="MySQL password. Defaults to 'testpass'.",
)
parser.addoption(
"--mysql-database",
dest="mysql_database",
default="test_db",
help="MySQL database name. Defaults to 'test_db'.",
)
parser.addoption(
"--mysql-host",
dest="mysql_host",
default="0.0.0.0",
help="Test against a MySQL server running on this host. Defaults to '0.0.0.0'.",
)
parser.addoption(
"--mysql-port",
dest="mysql_port",
type=int,
default=None,
help="The TCP port of the MySQL server.",
)
parser.addoption(
"--no-docker",
dest="use_docker",
default=True,
action="store_false",
help="Do not use a Docker MySQL image to run the tests. "
"If you decide to use this switch you will have to use a physical MySQL server.",
)
parser.addoption(
"--docker-mysql-image",
dest="docker_mysql_image",
default="mysql:latest",
help="Run the tests against a specific MySQL Docker image. Defaults to mysql:latest. "
"Check all supported versions here https://hub.docker.com/_/mysql",
)
@pytest.fixture(scope="session", autouse=True)
def cleanup_hanged_docker_containers() -> None:
try:
client: DockerClient = docker.from_env()
for container in client.containers.list():
if container.name == "pytest_mysql_to_sqlite3":
container.kill()
break
except Exception:
pass
def pytest_keyboard_interrupt() -> None:
try:
client: DockerClient = docker.from_env()
for container in client.containers.list():
if container.name == "pytest_mysql_to_sqlite3":
container.kill()
break
except Exception:
pass
class Helpers:
@staticmethod
@contextmanager
def not_raises(exception: t.Type[Exception]) -> t.Generator:
try:
yield
except exception:
raise pytest.fail(f"DID RAISE {exception}")
@staticmethod
@contextmanager
def session_scope(db: database.Database) -> t.Generator:
"""Provide a transactional scope around a series of operations."""
session: Session = db.Session()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
@pytest.fixture
def helpers() -> t.Type[Helpers]:
return Helpers
@pytest.fixture()
def sqlite_database(tmpdir: LocalPath) -> t.Union[str, Path, "os.PathLike[t.Any]"]:
db_name: str = "".join(choice(ascii_uppercase + ascii_lowercase + digits) for _ in range(32))
return Path(tmpdir.join(Path(f"{db_name}.sqlite3")))
def is_port_in_use(port: int, host: str = "0.0.0.0") -> bool:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
return s.connect_ex((host, port)) == 0
class MySQLCredentials(t.NamedTuple):
"""MySQL credentials."""
user: str
password: str
host: str
port: int
database: str
@pytest.fixture(scope="session")
def mysql_credentials(pytestconfig: Config) -> MySQLCredentials:
db_credentials_file: str = abspath(join(dirname(__file__), "db_credentials.json"))
if isfile(db_credentials_file):
with open(db_credentials_file, "r", "utf-8") as fh:
db_credentials: t.Dict[str, t.Any] = json.load(fh)
return MySQLCredentials(
user=db_credentials["mysql_user"],
password=db_credentials["mysql_password"],
database=db_credentials["mysql_database"],
host=db_credentials["mysql_host"],
port=db_credentials["mysql_port"],
)
port: int = pytestconfig.getoption("mysql_port") or 3306
if pytestconfig.getoption("use_docker"):
while is_port_in_use(port, pytestconfig.getoption("mysql_host")):
if port >= 2**16 - 1:
pytest.fail(f"No ports appear to be available on the host {pytestconfig.getoption('mysql_host')}")
port += 1
return MySQLCredentials(
user=pytestconfig.getoption("mysql_user") or "tester",
password=pytestconfig.getoption("mysql_password") or "testpass",
database=pytestconfig.getoption("mysql_database") or "test_db",
host=pytestconfig.getoption("mysql_host") or "0.0.0.0",
port=port,
)
@pytest.fixture(scope="session")
def mysql_instance(mysql_credentials: MySQLCredentials, pytestconfig: Config) -> t.Iterator[MySQLConnection]:
container: t.Optional[Container] = None
mysql_connection: t.Optional[t.Union[PooledMySQLConnection, MySQLConnection, CMySQLConnection]] = None
mysql_available: bool = False
mysql_connection_retries: int = 15 # failsafe
db_credentials_file = abspath(join(dirname(__file__), "db_credentials.json"))
if isfile(db_credentials_file):
use_docker = False
else:
use_docker = pytestconfig.getoption("use_docker")
if use_docker:
"""Connecting to a MySQL server within a Docker container is quite tricky :P
Read more on the issue here https://hub.docker.com/_/mysql#no-connections-until-mysql-init-completes
"""
try:
client = docker.from_env()
except Exception as err:
pytest.fail(str(err))
docker_mysql_image = pytestconfig.getoption("docker_mysql_image") or "mysql:latest"
if not any(docker_mysql_image in image.tags for image in client.images.list()):
print(f"Attempting to download Docker image {docker_mysql_image}'")
try:
client.images.pull(docker_mysql_image)
except (HTTPError, NotFound) as err:
pytest.fail(str(err))
container = client.containers.run(
image=docker_mysql_image,
name="pytest_mysql_to_sqlite3",
ports={"3306/tcp": (mysql_credentials.host, f"{mysql_credentials.port}/tcp")},
environment={
"MYSQL_RANDOM_ROOT_PASSWORD": "yes",
"MYSQL_USER": mysql_credentials.user,
"MYSQL_PASSWORD": mysql_credentials.password,
"MYSQL_DATABASE": mysql_credentials.database,
},
command=[
"--character-set-server=utf8mb4",
"--collation-server=utf8mb4_unicode_ci",
],
detach=True,
auto_remove=True,
)
while not mysql_available and mysql_connection_retries > 0:
try:
mysql_connection = mysql.connector.connect(
user=mysql_credentials.user,
password=mysql_credentials.password,
host=mysql_credentials.host,
port=mysql_credentials.port,
charset="utf8mb4",
collation="utf8mb4_unicode_ci",
)
except mysql.connector.Error as err:
if err.errno == errorcode.CR_SERVER_LOST:
# sleep for two seconds and retry the connection
sleep(2)
else:
raise
finally:
mysql_connection_retries -= 1
if mysql_connection and mysql_connection.is_connected():
mysql_available = True
mysql_connection.close()
else:
if not mysql_available and mysql_connection_retries <= 0:
raise ConnectionAbortedError("Maximum MySQL connection retries exhausted! Are you sure MySQL is running?")
yield # type: ignore[misc]
if use_docker and container is not None:
container.kill()
@pytest.fixture(scope="session")
def mysql_database(
tmpdir_factory: TempdirFactory,
mysql_instance: MySQLConnection,
mysql_credentials: MySQLCredentials,
_session_faker: Faker,
) -> t.Iterator[database.Database]:
temp_image_dir: LocalPath = tmpdir_factory.mktemp("images")
db: database.Database = database.Database(
f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
)
with Helpers.session_scope(db) as session:
for _ in range(_session_faker.pyint(min_value=12, max_value=24)):
article: models.Article = factories.ArticleFactory()
article.authors.append(factories.AuthorFactory())
article.tags.append(factories.TagFactory())
article.misc.append(factories.MiscFactory())
for _ in range(_session_faker.pyint(min_value=1, max_value=4)):
article.images.append(
factories.ImageFactory(
path=join(
str(temp_image_dir),
_session_faker.year(),
_session_faker.month(),
_session_faker.day_of_month(),
_session_faker.file_name(extension="jpg"),
)
)
)
session.add(article)
for _ in range(_session_faker.pyint(min_value=12, max_value=24)):
session.add(factories.CrazyNameFactory())
try:
session.commit()
except IntegrityError:
session.rollback()
yield db
if database_exists(db.engine.url):
drop_database(db.engine.url)
@pytest.fixture()
def cli_runner() -> t.Iterator[CliRunner]:
yield CliRunner()
0707010000002F000081A4000000000000000000000001682E58C100000597000000000000000000000000000000000000002900000000mysql-to-sqlite3-2.4.1/tests/database.pyimport typing as t
from datetime import datetime, timedelta
from decimal import Decimal
import simplejson as json
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy_utils import create_database, database_exists
from .models import Base
class Database:
engine: Engine
Session: sessionmaker
def __init__(self, database_uri):
self.Session = sessionmaker()
self.engine = create_engine(database_uri, json_serializer=self.dumps, json_deserializer=json.loads)
if not database_exists(self.engine.url):
create_database(self.engine.url)
self._create_db_tables()
self.Session.configure(bind=self.engine)
def _create_db_tables(self) -> None:
Base.metadata.create_all(self.engine)
@classmethod
def dumps(cls, data: t.Any) -> str:
return json.dumps(data, default=cls.json_serializer)
@staticmethod
def json_serializer(data: t.Any) -> t.Optional[str]:
if isinstance(data, datetime):
return data.isoformat()
if isinstance(data, Decimal):
return str(data)
if isinstance(data, timedelta):
hours, remainder = divmod(data.total_seconds(), 3600)
minutes, seconds = divmod(remainder, 60)
return "{:02}:{:02}:{:02}".format(int(hours), int(minutes), int(seconds))
return None
07070100000030000081A4000000000000000000000001682E58C1000011DF000000000000000000000000000000000000002A00000000mysql-to-sqlite3-2.4.1/tests/factories.pyimport typing as t
from os import environ
import factory
from . import faker_providers, models
factory.Faker.add_provider(faker_providers.DateTimeProviders)
class AuthorFactory(factory.Factory):
class Meta:
model: t.Type[models.Author] = models.Author
name: factory.Faker = factory.Faker("name")
class ImageFactory(factory.Factory):
class Meta:
model: t.Type[models.Image] = models.Image
path: factory.Faker = factory.Faker("file_path", depth=3, extension="jpg")
description: factory.Faker = factory.Faker("sentence", nb_words=12, variable_nb_words=True)
class TagFactory(factory.Factory):
class Meta:
model: t.Type[models.Tag] = models.Tag
name: factory.Faker = factory.Faker("sentence", nb_words=3, variable_nb_words=True)
class MiscFactory(factory.Factory):
class Meta:
model: t.Type[models.Misc] = models.Misc
big_integer_field: factory.Faker = factory.Faker("pyint", max_value=10**9)
large_binary_field: factory.Faker = factory.Faker("binary", length=1024 * 10)
boolean_field: factory.Faker = factory.Faker("boolean")
char_field: factory.Faker = factory.Faker("text", max_nb_chars=255)
date_field: factory.Faker = factory.Faker("date_this_decade")
date_time_field: factory.Faker = factory.Faker("date_time_this_century_without_microseconds")
decimal_field: factory.Faker = factory.Faker("pydecimal", left_digits=8, right_digits=2)
float_field: factory.Faker = factory.Faker("pyfloat", left_digits=8, right_digits=4)
integer_field: factory.Faker = factory.Faker("pyint", min_value=-(2**31), max_value=2**31 - 1)
if environ.get("LEGACY_DB", "0") == "0":
json_field: factory.Faker = factory.Faker("pydict")
nchar_field: factory.Faker = factory.Faker("text", max_nb_chars=255)
numeric_field: factory.Faker = factory.Faker("pyfloat", left_digits=8, right_digits=4)
unicode_field: factory.Faker = factory.Faker("text", max_nb_chars=255)
real_field: factory.Faker = factory.Faker("pyfloat", left_digits=8, right_digits=4)
small_integer_field: factory.Faker = factory.Faker("pyint", min_value=-(2**15), max_value=2**15 - 1)
string_field: factory.Faker = factory.Faker("text", max_nb_chars=255)
text_field: factory.Faker = factory.Faker("text", max_nb_chars=1024)
time_field: factory.Faker = factory.Faker("time_object_without_microseconds")
varbinary_field: factory.Faker = factory.Faker("binary", length=255)
varchar_field: factory.Faker = factory.Faker("text", max_nb_chars=255)
timestamp_field: factory.Faker = factory.Faker("date_time_this_century_without_microseconds")
class ArticleFactory(factory.Factory):
class Meta:
model: t.Type[models.Article] = models.Article
hash: factory.Faker = factory.Faker("md5")
title: factory.Faker = factory.Faker("sentence", nb_words=6)
slug: factory.Faker = factory.Faker("slug")
content: factory.Faker = factory.Faker("text", max_nb_chars=1024)
status: factory.Faker = factory.Faker("pystr", max_chars=1)
published: factory.Faker = factory.Faker("date_between", start_date="-1y", end_date="-1d")
@factory.post_generation
def authors(self, create, extracted, **kwargs):
if not create:
# Simple build, do nothing.
return
if extracted:
# A list of authors were passed in, use them
for author in extracted:
self.authors.add(author)
@factory.post_generation
def tags(self, create, extracted, **kwargs):
if not create:
# Simple build, do nothing.
return
if extracted:
# A list of authors were passed in, use them
for tag in extracted:
self.tags.add(tag)
@factory.post_generation
def images(self, create, extracted, **kwargs):
if not create:
# Simple build, do nothing.
return
if extracted:
# A list of authors were passed in, use them
for image in extracted:
self.images.add(image)
@factory.post_generation
def misc(self, create, extracted, **kwargs):
if not create:
# Simple build, do nothing.
return
if extracted:
# A list of authors were passed in, use them
for misc in extracted:
self.misc.add(misc)
class CrazyNameFactory(factory.Factory):
class Meta:
model: t.Type[models.CrazyName] = models.CrazyName
name: factory.Faker = factory.Faker("name")
07070100000031000081A4000000000000000000000001682E58C100000308000000000000000000000000000000000000003000000000mysql-to-sqlite3-2.4.1/tests/faker_providers.pyimport datetime
from typing import Optional
from faker.providers import BaseProvider, date_time
from faker.typing import DateParseType
class DateTimeProviders(BaseProvider):
def time_object_without_microseconds(self, end_datetime: Optional[DateParseType] = None) -> datetime.time:
return date_time.Provider(self.generator).time_object(end_datetime).replace(microsecond=0)
def date_time_this_century_without_microseconds(
self,
before_now: bool = True,
after_now: bool = False,
tzinfo: Optional[datetime.tzinfo] = None,
) -> datetime.datetime:
return (
date_time.Provider(self.generator)
.date_time_this_century(before_now, after_now, tzinfo)
.replace(microsecond=0)
)
07070100000032000041ED000000000000000000000002682E58C100000000000000000000000000000000000000000000002200000000mysql-to-sqlite3-2.4.1/tests/func07070100000033000081A4000000000000000000000001682E58C100000000000000000000000000000000000000000000002E00000000mysql-to-sqlite3-2.4.1/tests/func/__init__.py07070100000034000081A4000000000000000000000001682E58C10000BDCC000000000000000000000000000000000000003B00000000mysql-to-sqlite3-2.4.1/tests/func/mysql_to_sqlite3_test.pyimport logging
import os
import re
import typing as t
from collections import namedtuple
from decimal import Decimal
from pathlib import Path
from random import choice, sample
import mysql.connector
import pytest
import simplejson as json
from _pytest._py.path import LocalPath
from _pytest.logging import LogCaptureFixture
from faker import Faker
from mysql.connector import MySQLConnection, errorcode
from mysql.connector.connection_cext import CMySQLConnection
from mysql.connector.cursor import MySQLCursor
from mysql.connector.pooling import PooledMySQLConnection
from pytest_mock import MockFixture
from sqlalchemy import (
Connection,
CursorResult,
Engine,
Inspector,
MetaData,
Row,
Select,
Table,
TextClause,
create_engine,
inspect,
select,
text,
)
from sqlalchemy.engine.interfaces import ReflectedIndex
from mysql_to_sqlite3 import MySQLtoSQLite
from tests.conftest import Helpers, MySQLCredentials
from tests.database import Database
@pytest.mark.usefixtures("mysql_instance")
class TestMySQLtoSQLite:
@pytest.mark.init
@pytest.mark.parametrize(
"quiet",
[
pytest.param(False, id="verbose"),
pytest.param(True, id="quiet"),
],
)
def test_missing_mysql_user_raises_exception(self, mysql_credentials: MySQLCredentials, quiet: bool) -> None:
with pytest.raises(ValueError) as excinfo:
MySQLtoSQLite(mysql_database=mysql_credentials.database, quiet=quiet) # type: ignore[call-arg]
assert "Please provide a MySQL user" in str(excinfo.value)
@pytest.mark.init
@pytest.mark.parametrize(
"quiet",
[
pytest.param(False, id="verbose"),
pytest.param(True, id="quiet"),
],
)
def test_missing_mysql_database_raises_exception(self, faker: Faker, quiet: bool) -> None:
with pytest.raises(ValueError) as excinfo:
MySQLtoSQLite(mysql_user=faker.first_name().lower(), quiet=quiet) # type: ignore[call-arg]
assert "Please provide a MySQL database" in str(excinfo.value)
@pytest.mark.init
@pytest.mark.xfail
@pytest.mark.parametrize(
"quiet",
[
pytest.param(False, id="verbose"),
pytest.param(True, id="quiet"),
],
)
def test_invalid_mysql_credentials_raises_access_denied_exception(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
faker: Faker,
quiet: bool,
) -> None:
with pytest.raises(mysql.connector.Error) as excinfo:
MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=faker.first_name().lower(),
mysql_password=faker.password(length=16),
mysql_database=mysql_credentials.database,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
quiet=quiet,
)
assert "Access denied for user" in str(excinfo.value)
@pytest.mark.init
@pytest.mark.parametrize(
"quiet",
[
pytest.param(False, id="verbose"),
pytest.param(True, id="quiet"),
],
)
def test_bad_mysql_connection(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_credentials: MySQLCredentials,
mocker: MockFixture,
quiet: bool,
) -> None:
FakeConnector = namedtuple("FakeConnector", ["is_connected"])
mocker.patch.object(
mysql.connector,
"connect",
return_value=FakeConnector(is_connected=lambda: False),
)
with pytest.raises((ConnectionError, IOError)) as excinfo:
MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=mysql_credentials.user,
mysql_password=mysql_credentials.password,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
mysql_database=mysql_credentials.database,
chunk=1000,
quiet=quiet,
)
assert "Unable to connect to MySQL" in str(excinfo.value)
@pytest.mark.init
@pytest.mark.parametrize(
"exception, quiet",
[
pytest.param(
mysql.connector.Error(msg="Unknown database 'test_db'", errno=errorcode.ER_BAD_DB_ERROR),
False,
id="mysql.connector.Error verbose",
),
pytest.param(
mysql.connector.Error(msg="Unknown database 'test_db'", errno=errorcode.ER_BAD_DB_ERROR),
True,
id="mysql.connector.Error quiet",
),
pytest.param(Exception("Unknown database 'test_db'"), False, id="Exception verbose"),
pytest.param(Exception("Unknown database 'test_db'"), True, id="Exception quiet"),
],
)
def test_non_existing_mysql_database_raises_exception(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
faker: Faker,
mocker: MockFixture,
caplog: LogCaptureFixture,
exception: Exception,
quiet: bool,
) -> None:
class FakeMySQLConnection(MySQLConnection):
@property
def database(self) -> str:
return self._database
@database.setter
def database(self, value) -> None:
self._database = value
# raise a fake exception
raise exception
def is_connected(self) -> bool:
return True
def cursor(
self,
buffered: t.Optional[bool] = None,
raw: t.Optional[bool] = None,
prepared: t.Optional[bool] = None,
cursor_class: t.Optional[t.Type[MySQLCursor]] = None,
dictionary: t.Optional[bool] = None,
named_tuple: t.Optional[bool] = None,
) -> t.Union[t.Any, MySQLCursor]:
return True
caplog.set_level(logging.DEBUG)
mocker.patch.object(mysql.connector, "connect", return_value=FakeMySQLConnection())
with pytest.raises((mysql.connector.Error, Exception)) as excinfo:
MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=mysql_credentials.user,
mysql_password=mysql_credentials.password,
mysql_database=mysql_credentials.database,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
quiet=quiet,
)
assert any("MySQL Database does not exist!" in message for message in caplog.messages)
assert "Unknown database" in str(excinfo.value)
@pytest.mark.init
def test_without_tables_and_without_data(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
caplog: LogCaptureFixture,
tmpdir: LocalPath,
faker: Faker,
) -> None:
with pytest.raises(ValueError) as excinfo:
MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=mysql_credentials.user,
mysql_password=mysql_credentials.password,
mysql_database=mysql_credentials.database,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
without_tables=True,
without_data=True,
)
assert "Unable to continue without transferring data or creating tables!" in str(excinfo.value)
@pytest.mark.xfail
@pytest.mark.init
@pytest.mark.parametrize(
"quiet",
[
pytest.param(False, id="verbose"),
pytest.param(True, id="quiet"),
],
)
def test_log_to_file(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
caplog: LogCaptureFixture,
tmpdir: LocalPath,
faker: Faker,
quiet: bool,
) -> None:
log_file: LocalPath = tmpdir.join(Path("db.log"))
caplog.set_level(logging.DEBUG)
with pytest.raises(mysql.connector.Error):
MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=faker.first_name().lower(),
mysql_password=faker.password(length=16),
mysql_database=mysql_credentials.database,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
log_file=str(log_file),
quiet=quiet,
)
assert any("Access denied for user" in message for message in caplog.messages)
with log_file.open("r") as log_fh:
log = log_fh.read()
if len(caplog.messages) > 1:
assert caplog.messages[1] in log
else:
assert caplog.messages[0] in log
assert re.match(r"^\d{4,}-\d{2,}-\d{2,}\s+\d{2,}:\d{2,}:\d{2,}\s+\w+\s+", log) is not None
@pytest.mark.transfer
@pytest.mark.parametrize(
"chunk, vacuum, buffered, prefix_indices",
[
# 0000
pytest.param(
None,
False,
False,
False,
id="no chunk, no vacuum, no buffered cursor, no prefix indices",
),
# 0001
pytest.param(
None,
False,
False,
True,
id="no chunk, no vacuum, no buffered cursor, prefix indices",
),
# 1110
pytest.param(
10,
True,
True,
False,
id="chunk, vacuum, buffered cursor, no prefix indices",
),
# 1111
pytest.param(
10,
True,
True,
True,
id="chunk, vacuum, buffered cursor, prefix indices",
),
# 1100
pytest.param(
10,
True,
False,
False,
id="chunk, vacuum, no buffered cursor, no prefix indices",
),
# 1101
pytest.param(
10,
True,
False,
True,
id="chunk, vacuum, no buffered cursor, prefix indices",
),
# 0110
pytest.param(
None,
True,
True,
False,
id="no chunk, vacuum, buffered cursor, no prefix indices",
),
# 0111
pytest.param(
None,
True,
True,
True,
id="no chunk, vacuum, buffered cursor, prefix indices",
),
# 0100
pytest.param(
None,
True,
False,
False,
id="no chunk, vacuum, no buffered cursor, no prefix indices",
),
# 0101
pytest.param(
None,
True,
False,
True,
id="no chunk, vacuum, no buffered cursor, prefix indices",
),
# 1000
pytest.param(
10,
False,
False,
False,
id="chunk, no vacuum, no buffered cursor, no prefix indices",
),
# 1001
pytest.param(
10,
False,
False,
True,
id="chunk, no vacuum, no buffered cursor, prefix indices",
),
# 0010
pytest.param(
None,
False,
True,
False,
id="no chunk, no vacuum, buffered cursor, no prefix indices",
),
# 0011
pytest.param(
None,
False,
True,
True,
id="no chunk, no vacuum, buffered cursor, prefix indices",
),
# 1010
pytest.param(
10,
False,
True,
False,
id="chunk, no vacuum, buffered cursor, no prefix indices",
),
# 1011
pytest.param(
10,
False,
True,
True,
id="chunk, no vacuum, buffered cursor, prefix indices",
),
],
)
def test_transfer_transfers_all_tables_from_mysql_to_sqlite(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
helpers: Helpers,
caplog: LogCaptureFixture,
chunk: t.Optional[int],
vacuum: bool,
buffered: bool,
prefix_indices: bool,
) -> None:
proc: MySQLtoSQLite = MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=mysql_credentials.user,
mysql_password=mysql_credentials.password,
mysql_database=mysql_credentials.database,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
chunk=chunk,
vacuum=vacuum,
buffered=buffered,
prefix_indices=prefix_indices,
)
caplog.set_level(logging.DEBUG)
proc.transfer()
assert all(
message in [record.message for record in caplog.records]
for message in {
"Transferring table article_authors",
"Transferring table article_images",
"Transferring table article_tags",
"Transferring table articles",
"Transferring table authors",
"Transferring table images",
"Transferring table tags",
"Transferring table misc",
"Done!",
}
)
assert all(record.levelname == "INFO" for record in caplog.records)
assert not any(record.levelname == "ERROR" for record in caplog.records)
sqlite_engine: Engine = create_engine(
f"sqlite:///{sqlite_database}",
json_serializer=json.dumps,
json_deserializer=json.loads,
)
sqlite_cnx: Connection = sqlite_engine.connect()
sqlite_inspect: Inspector = inspect(sqlite_engine)
sqlite_tables: t.List[str] = sqlite_inspect.get_table_names()
mysql_engine: Engine = create_engine(
f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
)
mysql_cnx: Connection = mysql_engine.connect()
mysql_inspect: Inspector = inspect(mysql_engine)
mysql_tables: t.List[str] = mysql_inspect.get_table_names()
mysql_connector_connection: t.Union[PooledMySQLConnection, MySQLConnection, CMySQLConnection] = (
mysql.connector.connect(
user=mysql_credentials.user,
password=mysql_credentials.password,
host=mysql_credentials.host,
port=mysql_credentials.port,
database=mysql_credentials.database,
charset="utf8mb4",
collation="utf8mb4_unicode_ci",
)
)
server_version: t.Tuple[int, ...] = mysql_connector_connection.get_server_version()
""" Test if both databases have the same table names """
assert sqlite_tables == mysql_tables
""" Test if all the tables have the same column names """
for table_name in sqlite_tables:
assert [column["name"] for column in sqlite_inspect.get_columns(table_name)] == [
column["name"] for column in mysql_inspect.get_columns(table_name)
]
""" Test if all the tables have the same indices """
index_keys: t.Tuple[str, ...] = ("name", "column_names", "unique")
mysql_indices: t.List[ReflectedIndex] = []
for table_name in mysql_tables:
for index in mysql_inspect.get_indexes(table_name):
mysql_index: t.Dict[str, t.Any] = {}
for key in index_keys:
if key == "name" and prefix_indices:
mysql_index[key] = f"{table_name}_{index[key]}" # type: ignore[literal-required]
else:
mysql_index[key] = index[key] # type: ignore[literal-required]
mysql_indices.append(t.cast(ReflectedIndex, mysql_index))
for table_name in sqlite_tables:
for sqlite_index in sqlite_inspect.get_indexes(table_name):
sqlite_index["unique"] = bool(sqlite_index["unique"])
if "dialect_options" in sqlite_index:
sqlite_index.pop("dialect_options", None)
assert sqlite_index in mysql_indices
""" Test if all the tables have the same foreign keys """
for table_name in mysql_tables:
mysql_fk_stmt: TextClause = text(
"""
SELECT k.COLUMN_NAME AS `from`,
k.REFERENCED_TABLE_NAME AS `table`,
k.REFERENCED_COLUMN_NAME AS `to`,
c.UPDATE_RULE AS `on_update`,
c.DELETE_RULE AS `on_delete`
FROM information_schema.TABLE_CONSTRAINTS AS i
{JOIN} information_schema.KEY_COLUMN_USAGE AS k ON i.CONSTRAINT_NAME = k.CONSTRAINT_NAME
{JOIN} information_schema.REFERENTIAL_CONSTRAINTS c ON c.CONSTRAINT_NAME = i.CONSTRAINT_NAME
WHERE i.TABLE_SCHEMA = :table_schema
AND i.TABLE_NAME = :table_name
AND i.CONSTRAINT_TYPE = :constraint_type
""".format(
# MySQL 8.0.19 still works with "LEFT JOIN" everything above requires "JOIN"
JOIN="JOIN" if (server_version[0] == 8 and server_version[2] > 19) else "LEFT JOIN"
)
).bindparams(
table_schema=mysql_credentials.database,
table_name=table_name,
constraint_type="FOREIGN KEY",
)
mysql_fk_result: CursorResult = mysql_cnx.execute(mysql_fk_stmt)
mysql_foreign_keys: t.List[t.Dict[str, t.Any]] = [dict(row) for row in mysql_fk_result.mappings()]
sqlite_fk_stmt: TextClause = text(f'PRAGMA foreign_key_list("{table_name}")')
sqlite_fk_result: CursorResult = sqlite_cnx.execute(sqlite_fk_stmt)
if sqlite_fk_result.returns_rows:
for row in sqlite_fk_result.mappings():
fk: t.Dict[str, t.Any] = dict(row)
assert {
"table": fk["table"],
"from": fk["from"],
"to": fk["to"],
"on_update": fk["on_update"],
"on_delete": fk["on_delete"],
} in mysql_foreign_keys
""" Check if all the data was transferred correctly """
sqlite_results: t.List[t.Tuple[t.Tuple[t.Any, ...], ...]] = []
mysql_results: t.List[t.Tuple[t.Tuple[t.Any, ...], ...]] = []
meta: MetaData = MetaData()
for table_name in sqlite_tables:
sqlite_table: Table = Table(table_name, meta, autoload_with=sqlite_engine)
sqlite_stmt: Select = select(sqlite_table)
sqlite_result: t.List[Row[t.Any]] = list(sqlite_cnx.execute(sqlite_stmt).fetchall())
sqlite_result.sort()
sqlite_result_adapted: t.Tuple[t.Tuple[t.Any, ...], ...] = tuple(
tuple(float(data) if isinstance(data, Decimal) else data for data in row) for row in sqlite_result
)
sqlite_results.append(sqlite_result_adapted)
for table_name in mysql_tables:
mysql_table: Table = Table(table_name, meta, autoload_with=mysql_engine)
mysql_stmt: Select = select(mysql_table)
mysql_result: t.List[Row[t.Any]] = list(mysql_cnx.execute(mysql_stmt).fetchall())
mysql_result.sort()
mysql_result_adapted: t.Tuple[t.Tuple[t.Any, ...], ...] = tuple(
tuple(float(data) if isinstance(data, Decimal) else data for data in row) for row in mysql_result
)
mysql_results.append(mysql_result_adapted)
assert sqlite_results == mysql_results
mysql_cnx.close()
sqlite_cnx.close()
mysql_engine.dispose()
sqlite_engine.dispose()
@pytest.mark.transfer
def test_specific_tables_include_and_exclude_are_mutually_exclusive_options(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_credentials: MySQLCredentials,
caplog: LogCaptureFixture,
faker: Faker,
) -> None:
with pytest.raises(ValueError) as excinfo:
MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=mysql_credentials.user,
mysql_password=mysql_credentials.password,
mysql_database=mysql_credentials.database,
mysql_tables=faker.words(nb=3),
exclude_mysql_tables=faker.words(nb=3),
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
)
assert "mysql_tables and exclude_mysql_tables are mutually exclusive" in str(excinfo.value)
@pytest.mark.transfer
@pytest.mark.parametrize(
"chunk, vacuum, buffered, prefix_indices, exclude_tables",
[
# 00000
pytest.param(
None,
False,
False,
False,
False,
id="no chunk, no vacuum, no buffered cursor, no prefix indices, include tables",
),
# 00001
pytest.param(
None,
False,
False,
False,
True,
id="no chunk, no vacuum, no buffered cursor, no prefix indices, exclude tables",
),
# 00010
pytest.param(
None,
False,
False,
True,
False,
id="no chunk, no vacuum, no buffered cursor, prefix indices, include tables",
),
# 00011
pytest.param(
None,
False,
False,
True,
True,
id="no chunk, no vacuum, no buffered cursor, prefix indices, exclude tables",
),
# 11100
pytest.param(
10,
True,
True,
False,
False,
id="chunk, vacuum, buffered cursor, no prefix indices, include tables",
),
# 11101
pytest.param(
10,
True,
True,
False,
True,
id="chunk, vacuum, buffered cursor, no prefix indices, exclude tables",
),
# 11110
pytest.param(
10,
True,
True,
True,
False,
id="chunk, vacuum, buffered cursor, prefix indices, include tables",
),
# 11111
pytest.param(
10,
True,
True,
True,
True,
id="chunk, vacuum, buffered cursor, prefix indices, exclude tables",
),
# 11000
pytest.param(
10,
True,
False,
False,
False,
id="chunk, vacuum, no buffered cursor, no prefix indices, include tables",
),
# 11001
pytest.param(
10,
True,
False,
False,
True,
id="chunk, vacuum, no buffered cursor, no prefix indices, exclude tables",
),
# 11010
pytest.param(
10,
True,
False,
True,
False,
id="chunk, vacuum, no buffered cursor, prefix indices, include tables",
),
# 11011
pytest.param(
10,
True,
False,
True,
True,
id="chunk, vacuum, no buffered cursor, prefix indices, exclude tables",
),
# 01100
pytest.param(
None,
True,
True,
False,
False,
id="no chunk, vacuum, buffered cursor, no prefix indices, include tables",
),
# 01101
pytest.param(
None,
True,
True,
False,
True,
id="no chunk, vacuum, buffered cursor, no prefix indices, exclude tables",
),
# 01110
pytest.param(
None,
True,
True,
True,
False,
id="no chunk, vacuum, buffered cursor, prefix indices, include tables",
),
# 01111
pytest.param(
None,
True,
True,
True,
True,
id="no chunk, vacuum, buffered cursor, prefix indices, exclude tables",
),
# 01000
pytest.param(
None,
True,
False,
False,
False,
id="no chunk, vacuum, no buffered cursor, no prefix indices, include tables",
),
# 01001
pytest.param(
None,
True,
False,
False,
True,
id="no chunk, vacuum, no buffered cursor, no prefix indices, exclude tables",
),
# 01010
pytest.param(
None,
True,
False,
True,
False,
id="no chunk, vacuum, no buffered cursor, prefix indices, include tables",
),
# 01011
pytest.param(
None,
True,
False,
True,
True,
id="no chunk, vacuum, no buffered cursor, prefix indices, exclude tables",
),
# 10000
pytest.param(
10,
False,
False,
False,
False,
id="chunk, no vacuum, no buffered cursor, no prefix indices, include tables",
),
# 10001
pytest.param(
10,
False,
False,
False,
True,
id="chunk, no vacuum, no buffered cursor, no prefix indices, exclude tables",
),
# 10010
pytest.param(
10,
False,
False,
True,
False,
id="chunk, no vacuum, no buffered cursor, prefix indices, include tables",
),
# 10011
pytest.param(
10,
False,
False,
True,
True,
id="chunk, no vacuum, no buffered cursor, prefix indices, exclude tables",
),
# 00100
pytest.param(
None,
False,
True,
False,
False,
id="no chunk, no vacuum, buffered cursor, no prefix indices, include tables",
),
# 00101
pytest.param(
None,
False,
True,
False,
True,
id="no chunk, no vacuum, buffered cursor, no prefix indices, exclude tables",
),
# 00110
pytest.param(
None,
False,
True,
True,
False,
id="no chunk, no vacuum, buffered cursor, prefix indices, include tables",
),
# 00111
pytest.param(
None,
False,
True,
True,
True,
id="no chunk, no vacuum, buffered cursor, prefix indices, exclude tables",
),
# 10100
pytest.param(
10,
False,
True,
False,
False,
id="chunk, no vacuum, buffered cursor, no prefix indices, include tables",
),
# 10101
pytest.param(
10,
False,
True,
False,
True,
id="chunk, no vacuum, buffered cursor, no prefix indices, exclude tables",
),
# 10110
pytest.param(
10,
False,
True,
True,
False,
id="chunk, no vacuum, buffered cursor, prefix indices, include tables",
),
# 10111
pytest.param(
10,
False,
True,
True,
True,
id="chunk, no vacuum, buffered cursor, prefix indices, exclude tables",
),
],
)
def test_transfer_specific_tables_transfers_only_specified_tables_from_mysql_to_sqlite(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
helpers: Helpers,
caplog: LogCaptureFixture,
chunk: t.Optional[int],
vacuum: bool,
buffered: bool,
prefix_indices: bool,
exclude_tables: bool,
) -> None:
mysql_engine: Engine = create_engine(
f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
)
mysql_cnx: Connection = mysql_engine.connect()
mysql_inspect: Inspector = inspect(mysql_engine)
mysql_tables: t.List[str] = mysql_inspect.get_table_names()
table_number: int = choice(range(1, len(mysql_tables)))
random_mysql_tables: t.List[str] = sample(mysql_tables, table_number)
random_mysql_tables.sort()
remaining_tables: t.List[str] = list(set(mysql_tables) - set(random_mysql_tables))
remaining_tables.sort()
proc: MySQLtoSQLite = MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=mysql_credentials.user,
mysql_password=mysql_credentials.password,
mysql_database=mysql_credentials.database,
mysql_tables=None if exclude_tables else random_mysql_tables,
exclude_mysql_tables=random_mysql_tables if exclude_tables else None,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
prefix_indices=prefix_indices,
)
caplog.set_level(logging.DEBUG)
proc.transfer()
assert all(
message in [record.message for record in caplog.records]
for message in set(
[
f"Transferring table {table}"
for table in (remaining_tables if exclude_tables else random_mysql_tables)
]
+ ["Done!"]
)
)
assert all(record.levelname == "INFO" for record in caplog.records)
assert not any(record.levelname == "ERROR" for record in caplog.records)
sqlite_engine: Engine = create_engine(
f"sqlite:///{sqlite_database}",
json_serializer=json.dumps,
json_deserializer=json.loads,
)
sqlite_cnx: Connection = sqlite_engine.connect()
sqlite_inspect: Inspector = inspect(sqlite_engine)
sqlite_tables: t.List[str] = sqlite_inspect.get_table_names()
""" Test if both databases have the same table names """
if exclude_tables:
assert set(sqlite_tables) == set(remaining_tables)
else:
assert set(sqlite_tables) == set(random_mysql_tables)
""" Test if all the tables have the same column names """
for table_name in sqlite_tables:
assert [column["name"] for column in sqlite_inspect.get_columns(table_name)] == [
column["name"] for column in mysql_inspect.get_columns(table_name)
]
""" Test if all the tables have the same indices """
index_keys: t.Tuple[str, ...] = ("name", "column_names", "unique")
mysql_indices: t.List[ReflectedIndex] = []
for table_name in remaining_tables if exclude_tables else random_mysql_tables:
for index in mysql_inspect.get_indexes(table_name):
mysql_index: t.Dict[str, t.Any] = {}
for key in index_keys:
if key == "name" and prefix_indices:
mysql_index[key] = f"{table_name}_{index[key]}" # type: ignore[literal-required]
else:
mysql_index[key] = index[key] # type: ignore[literal-required]
mysql_indices.append(t.cast(ReflectedIndex, mysql_index))
for table_name in sqlite_tables:
for sqlite_index in sqlite_inspect.get_indexes(table_name):
sqlite_index["unique"] = bool(sqlite_index["unique"])
if "dialect_options" in sqlite_index:
sqlite_index.pop("dialect_options", None)
assert sqlite_index in mysql_indices
""" Check if all the data was transferred correctly """
sqlite_results: t.List[t.Tuple[t.Tuple[t.Any, ...], ...]] = []
mysql_results: t.List[t.Tuple[t.Tuple[t.Any, ...], ...]] = []
meta: MetaData = MetaData()
for table_name in sqlite_tables:
sqlite_table: Table = Table(table_name, meta, autoload_with=sqlite_engine)
sqlite_stmt: Select = select(sqlite_table)
sqlite_result: t.List[Row[t.Any]] = list(sqlite_cnx.execute(sqlite_stmt).fetchall())
sqlite_result.sort()
sqlite_result_adapted = tuple(
tuple(float(data) if isinstance(data, Decimal) else data for data in row) for row in sqlite_result
)
sqlite_results.append(sqlite_result_adapted)
for table_name in remaining_tables if exclude_tables else random_mysql_tables:
mysql_table: Table = Table(table_name, meta, autoload_with=mysql_engine)
mysql_stmt: Select = select(mysql_table)
mysql_result: t.List[Row[t.Any]] = list(mysql_cnx.execute(mysql_stmt).fetchall())
mysql_result.sort()
mysql_result_adapted = tuple(
tuple(float(data) if isinstance(data, Decimal) else data for data in row) for row in mysql_result
)
mysql_results.append(mysql_result_adapted)
assert sqlite_results == mysql_results
mysql_cnx.close()
sqlite_cnx.close()
mysql_engine.dispose()
sqlite_engine.dispose()
@pytest.mark.transfer
@pytest.mark.parametrize(
"chunk, vacuum, buffered, prefix_indices",
[
# 0000
pytest.param(
None,
False,
False,
False,
id="no chunk, no vacuum, no buffered cursor, no prefix indices",
),
# 0001
pytest.param(
None,
False,
False,
True,
id="no chunk, no vacuum, no buffered cursor, prefix indices",
),
# 1110
pytest.param(
10,
True,
True,
False,
id="chunk, vacuum, buffered cursor, no prefix indices",
),
# 1111
pytest.param(
10,
True,
True,
True,
id="chunk, vacuum, buffered cursor, prefix indices",
),
# 1100
pytest.param(
10,
True,
False,
False,
id="chunk, vacuum, no buffered cursor, no prefix indices",
),
# 1101
pytest.param(
10,
True,
False,
True,
id="chunk, vacuum, no buffered cursor, prefix indices",
),
# 0110
pytest.param(
None,
True,
True,
False,
id="no chunk, vacuum, buffered cursor, no prefix indices",
),
# 0111
pytest.param(
None,
True,
True,
True,
id="no chunk, vacuum, buffered cursor, prefix indices",
),
# 0100
pytest.param(
None,
True,
False,
False,
id="no chunk, vacuum, no buffered cursor, no prefix indices",
),
# 0101
pytest.param(
None,
True,
False,
True,
id="no chunk, vacuum, no buffered cursor, prefix indices",
),
# 1000
pytest.param(
10,
False,
False,
False,
id="chunk, no vacuum, no buffered cursor, no prefix indices",
),
# 1001
pytest.param(
10,
False,
False,
True,
id="chunk, no vacuum, no buffered cursor, prefix indices",
),
# 0010
pytest.param(
None,
False,
True,
False,
id="no chunk, no vacuum, buffered cursor, no prefix indices",
),
# 0011
pytest.param(
None,
False,
True,
True,
id="no chunk, no vacuum, buffered cursor, prefix indices",
),
# 1010
pytest.param(
10,
False,
True,
False,
id="chunk, no vacuum, buffered cursor, no prefix indices",
),
# 1011
pytest.param(
10,
False,
True,
True,
id="chunk, no vacuum, buffered cursor, prefix indices",
),
],
)
def test_transfer_limited_rows_from_mysql_to_sqlite(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
helpers: Helpers,
caplog: LogCaptureFixture,
chunk: t.Optional[int],
vacuum: bool,
buffered: bool,
prefix_indices: bool,
) -> None:
limit_rows: int = choice(range(1, 10))
proc: MySQLtoSQLite = MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=mysql_credentials.user,
mysql_password=mysql_credentials.password,
mysql_database=mysql_credentials.database,
limit_rows=limit_rows,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
prefix_indices=prefix_indices,
)
caplog.set_level(logging.DEBUG)
proc.transfer()
assert all(
message in [record.message for record in caplog.records]
for message in {
"Transferring table article_authors",
"Transferring table article_images",
"Transferring table article_tags",
"Transferring table articles",
"Transferring table authors",
"Transferring table images",
"Transferring table tags",
"Transferring table misc",
"Done!",
}
)
assert all(record.levelname == "INFO" for record in caplog.records)
assert not any(record.levelname == "ERROR" for record in caplog.records)
sqlite_engine: Engine = create_engine(
f"sqlite:///{sqlite_database}",
json_serializer=json.dumps,
json_deserializer=json.loads,
)
sqlite_cnx: Connection = sqlite_engine.connect()
sqlite_inspect: Inspector = inspect(sqlite_engine)
sqlite_tables: t.List[str] = sqlite_inspect.get_table_names()
mysql_engine: Engine = create_engine(
f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
)
mysql_cnx: Connection = mysql_engine.connect()
mysql_inspect: Inspector = inspect(mysql_engine)
mysql_tables: t.List[str] = mysql_inspect.get_table_names()
mysql_connector_connection: t.Union[PooledMySQLConnection, MySQLConnection, CMySQLConnection] = (
mysql.connector.connect(
user=mysql_credentials.user,
password=mysql_credentials.password,
host=mysql_credentials.host,
port=mysql_credentials.port,
database=mysql_credentials.database,
charset="utf8mb4",
collation="utf8mb4_unicode_ci",
)
)
server_version: t.Tuple[int, ...] = mysql_connector_connection.get_server_version()
""" Test if both databases have the same table names """
assert sqlite_tables == mysql_tables
""" Test if all the tables have the same column names """
for table_name in sqlite_tables:
assert [column["name"] for column in sqlite_inspect.get_columns(table_name)] == [
column["name"] for column in mysql_inspect.get_columns(table_name)
]
""" Test if all the tables have the same indices """
index_keys: t.Tuple[str, ...] = ("name", "column_names", "unique")
mysql_indices: t.List[ReflectedIndex] = []
for table_name in mysql_tables:
for index in mysql_inspect.get_indexes(table_name):
mysql_index: t.Dict[str, t.Any] = {}
for key in index_keys:
if key == "name" and prefix_indices:
mysql_index[key] = f"{table_name}_{index[key]}" # type: ignore[literal-required]
else:
mysql_index[key] = index[key] # type: ignore[literal-required]
mysql_indices.append(t.cast(ReflectedIndex, mysql_index))
for table_name in sqlite_tables:
for sqlite_index in sqlite_inspect.get_indexes(table_name):
sqlite_index["unique"] = bool(sqlite_index["unique"])
if "dialect_options" in sqlite_index:
sqlite_index.pop("dialect_options", None)
assert sqlite_index in mysql_indices
""" Test if all the tables have the same foreign keys """
for table_name in mysql_tables:
mysql_fk_stmt: TextClause = text(
"""
SELECT k.COLUMN_NAME AS `from`,
k.REFERENCED_TABLE_NAME AS `table`,
k.REFERENCED_COLUMN_NAME AS `to`,
c.UPDATE_RULE AS `on_update`,
c.DELETE_RULE AS `on_delete`
FROM information_schema.TABLE_CONSTRAINTS AS i
{JOIN} information_schema.KEY_COLUMN_USAGE AS k ON i.CONSTRAINT_NAME = k.CONSTRAINT_NAME
{JOIN} information_schema.REFERENTIAL_CONSTRAINTS c ON c.CONSTRAINT_NAME = i.CONSTRAINT_NAME
WHERE i.TABLE_SCHEMA = :table_schema
AND i.TABLE_NAME = :table_name
AND i.CONSTRAINT_TYPE = :constraint_type
""".format(
# MySQL 8.0.19 still works with "LEFT JOIN" everything above requires "JOIN"
JOIN="JOIN" if (server_version[0] == 8 and server_version[2] > 19) else "LEFT JOIN"
)
).bindparams(
table_schema=mysql_credentials.database,
table_name=table_name,
constraint_type="FOREIGN KEY",
)
mysql_fk_result: CursorResult = mysql_cnx.execute(mysql_fk_stmt)
mysql_foreign_keys: t.List[t.Dict[str, t.Any]] = [dict(row) for row in mysql_fk_result.mappings()]
sqlite_fk_stmt: TextClause = text(f'PRAGMA foreign_key_list("{table_name}")')
sqlite_fk_result: CursorResult = sqlite_cnx.execute(sqlite_fk_stmt)
if sqlite_fk_result.returns_rows:
for row in sqlite_fk_result.mappings():
fk: t.Dict[str, t.Any] = dict(row)
assert {
"table": fk["table"],
"from": fk["from"],
"to": fk["to"],
"on_update": fk["on_update"],
"on_delete": fk["on_delete"],
} in mysql_foreign_keys
""" Check if all the data was transferred correctly """
sqlite_results: t.List[t.Tuple[t.Tuple[t.Any, ...], ...]] = []
mysql_results: t.List[t.Tuple[t.Tuple[t.Any, ...], ...]] = []
meta: MetaData = MetaData()
for table_name in sqlite_tables:
sqlite_table: Table = Table(table_name, meta, autoload_with=sqlite_engine)
sqlite_stmt: Select = select(sqlite_table)
sqlite_result: t.List[Row[t.Any]] = list(sqlite_cnx.execute(sqlite_stmt).fetchall())
sqlite_result.sort()
sqlite_result_adapted = tuple(
tuple(float(data) if isinstance(data, Decimal) else data for data in row) for row in sqlite_result
)
sqlite_results.append(sqlite_result_adapted)
for table_name in mysql_tables:
mysql_table: Table = Table(table_name, meta, autoload_with=mysql_engine)
mysql_stmt: Select = select(mysql_table).limit(limit_rows)
mysql_result: t.List[Row[t.Any]] = list(mysql_cnx.execute(mysql_stmt).fetchall())
mysql_result.sort()
sqlite_result_adapted = tuple(
tuple(float(data) if isinstance(data, Decimal) else data for data in row) for row in mysql_result
)
mysql_results.append(sqlite_result_adapted)
assert sqlite_results == mysql_results
mysql_cnx.close()
sqlite_cnx.close()
mysql_engine.dispose()
sqlite_engine.dispose()
07070100000035000081A4000000000000000000000001682E58C1000054E1000000000000000000000000000000000000002E00000000mysql-to-sqlite3-2.4.1/tests/func/test_cli.pyimport os
import typing as t
from datetime import datetime
from random import choice, sample
import pytest
from click.testing import CliRunner, Result
from faker import Faker
from pytest_mock import MockFixture
from sqlalchemy import Connection, Engine, Inspector, create_engine, inspect
from mysql_to_sqlite3 import MySQLtoSQLite
from mysql_to_sqlite3 import __version__ as package_version
from mysql_to_sqlite3.cli import cli as mysql2sqlite
from tests.conftest import MySQLCredentials
from tests.database import Database
@pytest.mark.cli
@pytest.mark.usefixtures("mysql_instance")
class TestMySQLtoSQLite:
def test_no_arguments(self, cli_runner: CliRunner) -> None:
result: Result = cli_runner.invoke(mysql2sqlite)
assert result.exit_code in {0, 2}
assert all(
message in result.output
for message in {
f"Usage: {mysql2sqlite.name} [OPTIONS]",
f"{mysql2sqlite.name} version {package_version} Copyright (c) 2019-{datetime.now().year} Klemen Tusar",
}
)
def test_non_existing_sqlite_file(self, cli_runner: CliRunner, faker: Faker) -> None:
result: Result = cli_runner.invoke(mysql2sqlite, ["-f", faker.file_path(depth=1, extension=".sqlite3")])
assert result.exit_code > 0
assert any(
message in result.output
for message in {
'Error: Missing option "-d" / "--mysql-database"',
"Error: Missing option '-d' / '--mysql-database'",
}
)
def test_no_database_name(self, cli_runner: CliRunner, sqlite_database: "os.PathLike[t.Any]") -> None:
result: Result = cli_runner.invoke(mysql2sqlite, ["-f", str(sqlite_database)])
assert result.exit_code > 0
assert any(
message in result.output
for message in {
'Error: Missing option "-d" / "--mysql-database"',
"Error: Missing option '-d' / '--mysql-database'",
}
)
def test_no_database_user(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_credentials: MySQLCredentials,
) -> None:
result = cli_runner.invoke(mysql2sqlite, ["-f", str(sqlite_database), "-d", mysql_credentials.database])
assert result.exit_code > 0
assert any(
message in result.output
for message in {
'Error: Missing option "-u" / "--mysql-user"',
"Error: Missing option '-u' / '--mysql-user'",
}
)
@pytest.mark.xfail
def test_invalid_database_name(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
faker: Faker,
) -> None:
result: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
"_".join(faker.words(nb=3)),
"-u",
faker.first_name().lower(),
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
],
)
assert result.exit_code > 0
assert "1045 (28000): Access denied" in result.output
@pytest.mark.xfail
def test_invalid_database_user(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
faker: Faker,
) -> None:
result: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
faker.first_name().lower(),
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
],
)
assert result.exit_code > 0
assert "1045 (28000): Access denied" in result.output
@pytest.mark.xfail
def test_invalid_database_password(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
faker: Faker,
) -> None:
result: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"--mysql-password",
faker.password(length=16),
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
],
)
assert result.exit_code > 0
assert "1045 (28000): Access denied" in result.output
def test_database_password_prompt(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_credentials: MySQLCredentials,
mysql_database: Database,
) -> None:
result: Result = cli_runner.invoke(
mysql2sqlite,
args=[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"-p",
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
],
input=mysql_credentials.password,
)
assert result.exit_code == 0
@pytest.mark.xfail
def test_invalid_database_password_prompt(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_credentials: MySQLCredentials,
mysql_database: Database,
faker: Faker,
) -> None:
result: Result = cli_runner.invoke(
mysql2sqlite,
args=[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"-p",
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
],
input=faker.password(length=16),
)
assert result.exit_code > 0
assert "1045 (28000): Access denied" in result.output
@pytest.mark.xfail
def test_invalid_database_port(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
faker: Faker,
) -> None:
port: int = choice(range(2, 2**16 - 1))
if port == mysql_credentials.port:
port -= 1
result: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"--mysql-password",
mysql_credentials.password,
"-h",
mysql_credentials.host,
"-P",
str(port),
],
)
assert result.exit_code > 0
assert any(
message in result.output
for message in {
"2003 (HY000): Can't connect to MySQL server on",
"2003: Can't connect to MySQL server",
}
)
def test_without_data(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
) -> None:
result: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"--mysql-password",
mysql_credentials.password,
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
"-W",
],
)
assert result.exit_code == 0
def test_without_tables(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
) -> None:
# First we need to create the tables in the SQLite database
result1: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"--mysql-password",
mysql_credentials.password,
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
"-W",
],
)
assert result1.exit_code == 0
result2: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"--mysql-password",
mysql_credentials.password,
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
"-Z",
],
)
assert result2.exit_code == 0
@pytest.mark.parametrize(
"chunk, vacuum, use_buffered_cursors, quiet",
[
# 0000
pytest.param(
None,
False,
False,
False,
id="no chunk, no vacuum, no buffered cursor, verbose",
),
# 1110
pytest.param(10, True, True, False, id="chunk, vacuum, buffered cursor, verbose"),
# 1100
pytest.param(10, True, False, False, id="chunk, vacuum, no buffered cursor, verbose"),
# 0110
pytest.param(None, True, True, False, id="no chunk, vacuum, buffered cursor, verbose"),
# 0100
pytest.param(
None,
True,
False,
False,
id="no chunk, vacuum, no buffered cursor, verbose",
),
# 1000
pytest.param(
10,
False,
False,
False,
id="chunk, no vacuum, no buffered cursor, verbose",
),
# 0010
pytest.param(
None,
False,
True,
False,
id="no chunk, no vacuum, buffered cursor, verbose",
),
# 1010
pytest.param(10, False, True, False, id="chunk, no vacuum, buffered cursor, verbose"),
# 0001
pytest.param(
None,
False,
False,
True,
id="no chunk, no vacuum, no buffered cursor, quiet",
),
# 1111
pytest.param(10, True, True, True, id="chunk, vacuum, buffered cursor, quiet"),
# 1101
pytest.param(10, True, False, True, id="chunk, vacuum, no buffered cursor, quiet"),
# 0111
pytest.param(None, True, True, True, id="no chunk, vacuum, buffered cursor, quiet"),
# 0101
pytest.param(
None,
True,
False,
True,
id="no chunk, vacuum, no buffered cursor, quiet",
),
# 1001
pytest.param(10, False, False, True, id="chunk, no vacuum, no buffered cursor, quiet"),
# 0011
pytest.param(
None,
False,
True,
True,
id="no chunk, no vacuum, buffered cursor, quiet",
),
# 1011
pytest.param(10, False, True, True, id="chunk, no vacuum, buffered cursor, quiet"),
],
)
def test_minimum_valid_parameters(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
chunk: t.Optional[int],
vacuum: bool,
use_buffered_cursors: bool,
quiet: bool,
) -> None:
arguments: t.List[str] = [
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"--mysql-password",
mysql_credentials.password,
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
]
if chunk:
arguments.append("-c")
arguments.append(str(chunk))
if vacuum:
arguments.append("-V")
if use_buffered_cursors:
arguments.append("--use-buffered-cursors")
if quiet:
arguments.append("-q")
result: Result = cli_runner.invoke(mysql2sqlite, arguments)
assert result.exit_code == 0
copyright_header = (
f"{mysql2sqlite.name} version {package_version} Copyright (c) 2019-{datetime.now().year} Klemen Tusar\n"
)
assert copyright_header in result.output
if quiet:
assert result.output.replace(copyright_header, "") == ""
else:
assert result.output.replace(copyright_header, "") != ""
def test_keyboard_interrupt(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_credentials: MySQLCredentials,
mysql_database: Database,
mocker: MockFixture,
) -> None:
mocker.patch.object(MySQLtoSQLite, "transfer", side_effect=KeyboardInterrupt())
result: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"--mysql-password",
mysql_credentials.password,
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
],
)
assert result.exit_code > 0
assert "Process interrupted" in result.output
def test_specific_tables_include_and_exclude_are_mutually_exclusive_options(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_credentials: MySQLCredentials,
mysql_database: Database,
) -> None:
mysql_engine: Engine = create_engine(
f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
)
mysql_cnx: Connection = mysql_engine.connect()
mysql_inspect: Inspector = inspect(mysql_engine)
mysql_tables: t.List[str] = mysql_inspect.get_table_names()
table_number: int = choice(range(1, len(mysql_tables) // 2))
include_mysql_tables: t.List[str] = sample(mysql_tables, table_number)
include_mysql_tables.sort()
exclude_mysql_tables = list(set(sample(mysql_tables, table_number)) - set(include_mysql_tables))
exclude_mysql_tables.sort()
result: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-t",
" ".join(include_mysql_tables),
"-e",
" ".join(exclude_mysql_tables),
"-u",
mysql_credentials.user,
"--mysql-password",
mysql_credentials.password,
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
],
)
assert result.exit_code > 0
assert "Illegal usage: --mysql-tables and --exclude-mysql-tables are mutually exclusive!" in result.output
mysql_cnx.close()
mysql_engine.dispose()
def test_transfer_specific_tables_only(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_credentials: MySQLCredentials,
mysql_database: Database,
) -> None:
mysql_engine: Engine = create_engine(
f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
)
mysql_inspect: Inspector = inspect(mysql_engine)
mysql_tables: t.List[str] = mysql_inspect.get_table_names()
table_number: int = choice(range(1, len(mysql_tables)))
result: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-t",
" ".join(sample(mysql_tables, table_number)),
"-u",
mysql_credentials.user,
"--mysql-password",
mysql_credentials.password,
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
],
)
assert result.exit_code == 0
@pytest.mark.xfail
def test_version(self, cli_runner: CliRunner) -> None:
result = cli_runner.invoke(mysql2sqlite, ["--version"])
assert result.exit_code == 0
assert all(
message in result.output
for message in {
"mysql-to-sqlite3",
"Operating",
"System",
"Python",
"MySQL",
"SQLite",
"click",
"mysql-connector-python",
"python-slugify",
"pytimeparse2",
"simplejson",
"tabulate",
"tqdm",
}
)
def test_invalid_mysql_charset_collation(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_credentials: MySQLCredentials,
mysql_database: Database,
) -> None:
"""Test CLI with invalid collation for the specified charset."""
result: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"--mysql-password",
mysql_credentials.password,
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
"--mysql-charset",
"utf8mb4",
"--mysql-collation",
"invalid_collation",
],
)
assert result.exit_code > 0
assert "Error: Invalid value for '--mysql-collation': 'invalid_collation'" in result.output
def test_without_tables_and_without_data_flags(
self,
cli_runner: CliRunner,
sqlite_database: "os.PathLike[t.Any]",
mysql_credentials: MySQLCredentials,
) -> None:
"""Test CLI with both --without-tables and --without-data flags set."""
result: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"--mysql-password",
mysql_credentials.password,
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
"--without-tables",
"--without-data",
],
)
assert result.exit_code > 0
assert (
"Error: Both -Z/--without-tables and -W/--without-data are set. There is nothing to do. Exiting..."
in result.output
)
def test_passwordless_login(
self, cli_runner: CliRunner, sqlite_database: "os.PathLike[t.Any]", mysql_credentials: MySQLCredentials
) -> None:
result: Result = cli_runner.invoke(
mysql2sqlite,
[
"-f",
str(sqlite_database),
"-d",
mysql_credentials.database,
"-u",
mysql_credentials.user,
"-h",
mysql_credentials.host,
"-P",
str(mysql_credentials.port),
],
)
assert "using password: NO" in result.output
07070100000036000081A4000000000000000000000001682E58C100001B0C000000000000000000000000000000000000002700000000mysql-to-sqlite3-2.4.1/tests/models.pyimport typing as t
from datetime import date, datetime, time
from decimal import Decimal
from os import environ
from sqlalchemy import (
CHAR,
DECIMAL,
JSON,
NCHAR,
REAL,
TIMESTAMP,
VARBINARY,
VARCHAR,
BigInteger,
Column,
ForeignKey,
Integer,
LargeBinary,
Numeric,
SmallInteger,
String,
Table,
Text,
Time,
Unicode,
)
from sqlalchemy.dialects.mysql import BIGINT, INTEGER, MEDIUMINT, SMALLINT, TINYINT
from sqlalchemy.orm import DeclarativeBase, Mapped, backref, mapped_column, relationship
from sqlalchemy.sql.functions import current_timestamp
class Base(DeclarativeBase):
pass
class Author(Base):
__tablename__ = "authors"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
dupe: Mapped[bool] = mapped_column(index=True, default=False)
def __repr__(self):
return f"<Author(id='{self.id}', name='{self.name}')>"
article_authors = Table(
"article_authors",
Base.metadata,
Column("article_id", Integer, ForeignKey("articles.id"), primary_key=True),
Column("author_id", Integer, ForeignKey("authors.id"), primary_key=True),
)
class Image(Base):
__tablename__ = "images"
id: Mapped[int] = mapped_column(primary_key=True)
path: Mapped[str] = mapped_column(String(255), index=True)
description: Mapped[str] = mapped_column(String(255), nullable=True)
dupe: Mapped[bool] = mapped_column(index=True, default=False)
def __repr__(self):
return f"<Image(id='{self.id}', path='{self.path}')>"
article_images = Table(
"article_images",
Base.metadata,
Column("article_id", Integer, ForeignKey("articles.id"), primary_key=True),
Column("image_id", Integer, ForeignKey("images.id"), primary_key=True),
)
class Tag(Base):
__tablename__ = "tags"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
dupe: Mapped[bool] = mapped_column(index=True, default=False)
def __repr__(self):
return f"<Tag(id='{self.id}', name='{self.name}')>"
article_tags = Table(
"article_tags",
Base.metadata,
Column("article_id", Integer, ForeignKey("articles.id"), primary_key=True),
Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True),
)
class Misc(Base):
"""This model contains all possible MySQL types"""
__tablename__ = "misc"
id: Mapped[int] = mapped_column(primary_key=True)
big_integer_field: Mapped[int] = mapped_column(BigInteger, default=0)
big_integer_unsigned_field: Mapped[int] = mapped_column(BIGINT(unsigned=True), default=0)
if environ.get("LEGACY_DB", "0") == "0":
large_binary_field: Mapped[bytes] = mapped_column(LargeBinary, nullable=True, default=b"Lorem ipsum dolor")
else:
large_binary_field = mapped_column(LargeBinary, nullable=True)
boolean_field: Mapped[bool] = mapped_column(default=False)
char_field: Mapped[str] = mapped_column(CHAR(255), nullable=True)
date_field: Mapped[date] = mapped_column(nullable=True)
date_time_field: Mapped[datetime] = mapped_column(nullable=True)
decimal_field: Mapped[Decimal] = mapped_column(DECIMAL(10, 2), nullable=True)
float_field: Mapped[Decimal] = mapped_column(DECIMAL(12, 4), default=0)
integer_field: Mapped[int] = mapped_column(default=0)
integer_unsigned_field: Mapped[int] = mapped_column(INTEGER(unsigned=True), default=0)
tinyint_field: Mapped[int] = mapped_column(TINYINT, default=0)
tinyint_unsigned_field: Mapped[int] = mapped_column(TINYINT(unsigned=True), default=0)
mediumint_field: Mapped[int] = mapped_column(MEDIUMINT, default=0)
mediumint_unsigned_field: Mapped[int] = mapped_column(MEDIUMINT(unsigned=True), default=0)
if environ.get("LEGACY_DB", "0") == "0":
json_field: Mapped[t.Mapping[str, t.Any]] = mapped_column(JSON, nullable=True)
nchar_field: Mapped[str] = mapped_column(NCHAR(255), nullable=True)
numeric_field: Mapped[float] = mapped_column(Numeric(12, 4), default=0)
unicode_field: Mapped[str] = mapped_column(Unicode(255), nullable=True)
real_field: Mapped[float] = mapped_column(REAL(12), default=0)
small_integer_field: Mapped[int] = mapped_column(SmallInteger, default=0)
small_integer_unsigned_field: Mapped[int] = mapped_column(SMALLINT(unsigned=True), default=0)
string_field: Mapped[str] = mapped_column(String(255), nullable=True)
text_field: Mapped[str] = mapped_column(Text, nullable=True)
time_field: Mapped[time] = mapped_column(Time, nullable=True)
varbinary_field: Mapped[bytes] = mapped_column(VARBINARY(255), nullable=True)
varchar_field: Mapped[str] = mapped_column(VARCHAR(255), nullable=True)
timestamp_field: Mapped[datetime] = mapped_column(TIMESTAMP, default=current_timestamp())
dupe: Mapped[bool] = mapped_column(index=True, default=False)
article_misc = Table(
"article_misc",
Base.metadata,
Column("article_id", Integer, ForeignKey("articles.id"), primary_key=True),
Column("misc_id", Integer, ForeignKey("misc.id"), primary_key=True),
)
class Article(Base):
__tablename__ = "articles"
id: Mapped[int] = mapped_column(primary_key=True)
hash: Mapped[str] = mapped_column(String(32), unique=True)
slug: Mapped[str] = mapped_column(String(255), index=True)
title: Mapped[str] = mapped_column(String(255), index=True)
content: Mapped[str] = mapped_column(Text, nullable=True)
status: Mapped[str] = mapped_column(CHAR(1), index=True)
published: Mapped[datetime] = mapped_column(nullable=True)
dupe: Mapped[bool] = mapped_column(index=True, default=False)
# relationships
authors: Mapped[t.List[Author]] = relationship(
"Author",
secondary=article_authors,
backref=backref("authors", lazy="dynamic"),
lazy="dynamic",
)
tags: Mapped[t.List[Tag]] = relationship(
"Tag",
secondary=article_tags,
backref=backref("tags", lazy="dynamic"),
lazy="dynamic",
)
images: Mapped[t.List[Image]] = relationship(
"Image",
secondary=article_images,
backref=backref("images", lazy="dynamic"),
lazy="dynamic",
)
misc: Mapped[t.List[Misc]] = relationship(
"Misc",
secondary=article_misc,
backref=backref("misc", lazy="dynamic"),
lazy="dynamic",
)
def __repr__(self):
return f"<Article(id='{self.id}', title='{self.title}')>"
class CrazyName(Base):
__tablename__ = "crazy_name."
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
dupe: Mapped[bool] = mapped_column(index=True, default=False)
def __repr__(self):
return f"<CrazyName(id='{self.id}', name='{self.name}')>"
07070100000037000041ED000000000000000000000002682E58C100000000000000000000000000000000000000000000002200000000mysql-to-sqlite3-2.4.1/tests/unit07070100000038000081A4000000000000000000000001682E58C100000000000000000000000000000000000000000000002E00000000mysql-to-sqlite3-2.4.1/tests/unit/__init__.py07070100000039000081A4000000000000000000000001682E58C10000639C000000000000000000000000000000000000003B00000000mysql-to-sqlite3-2.4.1/tests/unit/mysql_to_sqlite3_test.pyimport logging
import os
import sqlite3
import typing as t
from random import choice
import mysql.connector
import pytest
from _pytest.logging import LogCaptureFixture
from mysql.connector import errorcode
from pytest_mock import MockerFixture, MockFixture
from sqlalchemy import Inspector, inspect
from sqlalchemy.dialects.mysql import __all__ as mysql_column_types
from mysql_to_sqlite3 import MySQLtoSQLite
from mysql_to_sqlite3.sqlite_utils import CollatingSequences
from tests.conftest import MySQLCredentials
from tests.database import Database
class TestMySQLtoSQLiteClassmethods:
def test_translate_type_from_mysql_to_sqlite_invalid_column_type(
self,
mocker: MockFixture,
) -> None:
with pytest.raises(ValueError) as excinfo:
mocker.patch.object(MySQLtoSQLite, "_valid_column_type", return_value=False)
MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type="text")
assert "is not a valid column_type!" in str(excinfo.value)
def test_translate_type_from_mysql_to_sqlite_all_valid_columns(self) -> None:
for column_type in mysql_column_types + (
"BIGINT UNSIGNED",
"INTEGER UNSIGNED",
"INT",
"INT UNSIGNED",
"SMALLINT UNSIGNED",
"TINYINT UNSIGNED",
"MEDIUMINT UNSIGNED",
"CHAR(2)",
"NCHAR(7)",
"NVARCHAR(17)",
"VARCHAR(123)",
):
if any(c for c in column_type if c.islower()):
continue
elif column_type == "INT":
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "INTEGER"
elif column_type == "DECIMAL":
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "DECIMAL"
elif column_type == "YEAR":
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "YEAR"
elif column_type == "TIME":
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "TIME"
elif column_type == "TIMESTAMP":
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "DATETIME"
elif column_type in {
"BINARY",
"BIT",
"LONGBLOB",
"MEDIUMBLOB",
"TINYBLOB",
"VARBINARY",
}:
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "BLOB"
elif column_type == "CHAR":
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "CHARACTER"
elif column_type == "CHAR(2)":
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "CHARACTER(2)"
elif column_type == "NCHAR(7)":
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "NCHAR(7)"
elif column_type == "NVARCHAR(17)":
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "NVARCHAR(17)"
elif column_type == "VARCHAR(123)":
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "VARCHAR(123)"
elif column_type in {
"ENUM",
"LONGTEXT",
"MEDIUMTEXT",
"SET",
"TINYTEXT",
"INET4",
"INET6",
}:
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "TEXT"
elif column_type == "JSON":
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "TEXT"
assert (
MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type, sqlite_json1_extension_enabled=True)
== "JSON"
)
elif column_type.endswith(" UNSIGNED"):
if column_type.startswith("INT "):
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == "INTEGER"
else:
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == column_type.replace(
" UNSIGNED", ""
)
else:
assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite(column_type) == column_type
@pytest.mark.parametrize(
"column_default, column_extra, sqlite_default_translation",
[
pytest.param(None, None, "", id="None"),
pytest.param("", None, "DEFAULT ''", id='""'),
pytest.param("lorem", None, "DEFAULT 'lorem'", id='"lorem"'),
pytest.param(
"lorem ipsum dolor",
None,
"DEFAULT 'lorem ipsum dolor'",
id='"lorem ipsum dolor"',
),
pytest.param("CURRENT_TIME", "DEFAULT_GENERATED", "DEFAULT CURRENT_TIME", id='"CURRENT_TIME"'),
pytest.param("current_time", "DEFAULT_GENERATED", "DEFAULT CURRENT_TIME", id='"current_time"'),
pytest.param("CURRENT_DATE", "DEFAULT_GENERATED", "DEFAULT CURRENT_DATE", id='"CURRENT_DATE"'),
pytest.param("current_date", "DEFAULT_GENERATED", "DEFAULT CURRENT_DATE", id='"current_date"'),
pytest.param(
"CURRENT_TIMESTAMP",
"DEFAULT_GENERATED",
"DEFAULT CURRENT_TIMESTAMP",
id='"CURRENT_TIMESTAMP"',
),
pytest.param(
"current_timestamp",
"DEFAULT_GENERATED",
"DEFAULT CURRENT_TIMESTAMP",
id='"current_timestamp"',
),
pytest.param(r"""_utf8mb4\'[]\'""", "DEFAULT_GENERATED", "DEFAULT '[]'", id=r"""_utf8mb4\'[]\'"""),
pytest.param(r"""_latin1\'abc\'""", "DEFAULT_GENERATED", "DEFAULT 'abc'", id=r"""_latin1\'abc\'"""),
pytest.param(r"""_binary\'abc\'""", "DEFAULT_GENERATED", "DEFAULT 'abc'", id=r"""_binary\'abc\'"""),
pytest.param(
r"""_latin1 X\'4D7953514C\'""",
"DEFAULT_GENERATED",
"DEFAULT x'4D7953514C'",
id=r"""_latin1 X\'4D7953514C\'""",
),
pytest.param(
r"""_latin1 b\'1000001\'""", "DEFAULT_GENERATED", "DEFAULT 'A'", id=r"""_latin1 b\'1000001\'"""
),
],
)
def test_translate_default_from_mysql_to_sqlite(
self,
column_default: t.Optional[str],
column_extra: t.Optional[str],
sqlite_default_translation: str,
) -> None:
assert (
MySQLtoSQLite._translate_default_from_mysql_to_sqlite(column_default, column_extra=column_extra)
== sqlite_default_translation
)
@pytest.mark.parametrize(
"column_default, sqlite_default_translation, sqlite_version",
[
pytest.param(False, "DEFAULT(FALSE)", "3.23.0", id="False (NEW)"),
pytest.param(True, "DEFAULT(TRUE)", "3.23.0", id="True (NEW)"),
pytest.param(False, "DEFAULT '0'", "3.22.0", id="False (OLD)"),
pytest.param(True, "DEFAULT '1'", "3.22.0", id="True (OLD)"),
],
)
def test_translate_default_booleans_from_mysql_to_sqlite(
self,
mocker: MockerFixture,
column_default: bool,
sqlite_default_translation: str,
sqlite_version: str,
) -> None:
mocker.patch.object(sqlite3, "sqlite_version", sqlite_version)
assert (
MySQLtoSQLite._translate_default_from_mysql_to_sqlite(column_default, "BOOLEAN")
== sqlite_default_translation
)
@pytest.mark.parametrize(
"column_default, sqlite_default_translation, column_type",
[
pytest.param("0", "DEFAULT '0'", "NUMERIC", id='"0" (NUMERIC)'),
pytest.param("1", "DEFAULT '1'", "NUMERIC", id='"1" (NUMERIC)'),
pytest.param("0", "DEFAULT '0'", "TEXT", id='"0" (TEXT)'),
pytest.param("1", "DEFAULT '1'", "TEXT", id='"1" (TEXT)'),
pytest.param(0, "DEFAULT '0'", "NUMERIC", id="0 (NUMERIC)"),
pytest.param(1, "DEFAULT '1'", "NUMERIC", id="1 (NUMERIC)"),
pytest.param(0, "DEFAULT '0'", "TEXT", id="0 (TEXT)"),
pytest.param(1, "DEFAULT '1'", "TEXT", id="1 (TEXT)"),
pytest.param(123456789, "DEFAULT '123456789'", "NUMERIC", id="123456789 (NUMERIC)"),
pytest.param(1234.56789, "DEFAULT '1234.56789'", "NUMERIC", id="1234.56789 (NUMERIC)"),
pytest.param(123456789, "DEFAULT '123456789'", "TEXT", id="123456789 (TEXT)"),
pytest.param(1234.56789, "DEFAULT '1234.56789'", "TEXT", id="1234.56789 (TEXT)"),
],
)
def test_translate_default_numbers_from_mysql_to_sqlite(
self,
column_default: t.Union[int, float, str],
sqlite_default_translation: str,
column_type: str,
) -> None:
assert (
MySQLtoSQLite._translate_default_from_mysql_to_sqlite(column_default, column_type)
== sqlite_default_translation
)
@pytest.mark.parametrize(
"column_default, sqlite_default_translation",
[
pytest.param(b"", "DEFAULT x''", id="b''"),
pytest.param(b"-1", "DEFAULT x'2d31'", id="b'-1'"),
pytest.param(b"0", "DEFAULT x'30'", id="b'0'"),
pytest.param(b"1", "DEFAULT x'31'", id="b'1'"),
pytest.param(b"-1234567890", "DEFAULT x'2d31323334353637383930'", id="b'-1234567890'"),
pytest.param(b"1234567890", "DEFAULT x'31323334353637383930'", id="b'1234567890'"),
pytest.param(b"SQLite", "DEFAULT x'53514c697465'", id="b'SQLite'"),
pytest.param(
b"Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nam pretium, purus vitae sollicitudin varius, nisi lectus vehicula dui, ut dignissim felis dolor blandit justo. Donec eleifend lectus ut feugiat rhoncus. Donec erat nibh, dapibus nec diam id, lacinia lacinia nisl. Mauris sagittis efficitur nisl. Ut tincidunt elementum rhoncus. Cras suscipit dolor sed est ultricies, quis dapibus neque suscipit. Etiam ac enim eu ligula bibendum blandit quis sit amet felis. Praesent mi nisi, luctus sit amet nunc ut, fermentum tempus purus. Suspendisse vel purus a nibh aliquam hendrerit. Aliquam sit amet tristique lorem. Sed elementum congue ante id mollis. Donec vitae pretium neque.",
"DEFAULT x'4c6f72656d20697073756d20646f6c6f722073697420616d65742c20636f6e73656374657475722061646970697363696e6720656c69742e204e616d207072657469756d2c20707572757320766974616520736f6c6c696369747564696e207661726975732c206e697369206c6563747573207665686963756c61206475692c207574206469676e697373696d2066656c697320646f6c6f7220626c616e646974206a7573746f2e20446f6e656320656c656966656e64206c656374757320757420666575676961742072686f6e6375732e20446f6e65632065726174206e6962682c2064617069627573206e6563206469616d2069642c206c6163696e6961206c6163696e6961206e69736c2e204d617572697320736167697474697320656666696369747572206e69736c2e2055742074696e636964756e7420656c656d656e74756d2072686f6e6375732e204372617320737573636970697420646f6c6f72207365642065737420756c747269636965732c20717569732064617069627573206e657175652073757363697069742e20457469616d20616320656e696d206575206c6967756c6120626962656e64756d20626c616e64697420717569732073697420616d65742066656c69732e205072616573656e74206d69206e6973692c206c75637475732073697420616d6574206e756e632075742c206665726d656e74756d2074656d7075732070757275732e2053757370656e64697373652076656c2070757275732061206e69626820616c697175616d2068656e6472657269742e20416c697175616d2073697420616d657420747269737469717565206c6f72656d2e2053656420656c656d656e74756d20636f6e67756520616e7465206964206d6f6c6c69732e20446f6e6563207669746165207072657469756d206e657175652e'",
id="b'Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nam pretium, purus vitae sollicitudin varius, nisi lectus vehicula dui, ut dignissim felis dolor blandit justo. Donec eleifend lectus ut feugiat rhoncus. Donec erat nibh, dapibus nec diam id, lacinia lacinia nisl. Mauris sagittis efficitur nisl. Ut tincidunt elementum rhoncus. Cras suscipit dolor sed est ultricies, quis dapibus neque suscipit. Etiam ac enim eu ligula bibendum blandit quis sit amet felis. Praesent mi nisi, luctus sit amet nunc ut, fermentum tempus purus. Suspendisse vel purus a nibh aliquam hendrerit. Aliquam sit amet tristique lorem. Sed elementum congue ante id mollis. Donec vitae pretium neque.'",
),
],
)
def test_translate_default_blob_bytes_from_mysql_to_sqlite(
self,
column_default: bytes,
sqlite_default_translation: str,
) -> None:
assert (
MySQLtoSQLite._translate_default_from_mysql_to_sqlite(column_default, "BLOB") == sqlite_default_translation
)
@pytest.mark.parametrize(
"collation, resulting_column_collation, column_type",
[
pytest.param(
CollatingSequences.BINARY,
"",
"CHARACTER",
id=f"{CollatingSequences.BINARY} (CHARACTER)",
),
pytest.param(
CollatingSequences.NOCASE,
f"COLLATE {CollatingSequences.NOCASE}",
"CHARACTER",
id=f"{CollatingSequences.NOCASE} (CHARACTER)",
),
pytest.param(
CollatingSequences.RTRIM,
f"COLLATE {CollatingSequences.RTRIM}",
"CHARACTER",
id=f"{CollatingSequences.RTRIM} (CHARACTER)",
),
pytest.param(
CollatingSequences.BINARY,
"",
"NCHAR",
id=f"{CollatingSequences.BINARY} (NCHAR)",
),
pytest.param(
CollatingSequences.NOCASE,
f"COLLATE {CollatingSequences.NOCASE}",
"NCHAR",
id=f"{CollatingSequences.NOCASE} (NCHAR)",
),
pytest.param(
CollatingSequences.RTRIM,
f"COLLATE {CollatingSequences.RTRIM}",
"NCHAR",
id=f"{CollatingSequences.RTRIM} (NCHAR)",
),
pytest.param(
CollatingSequences.BINARY,
"",
"NVARCHAR",
id=f"{CollatingSequences.BINARY} (NVARCHAR)",
),
pytest.param(
CollatingSequences.NOCASE,
f"COLLATE {CollatingSequences.NOCASE}",
"NVARCHAR",
id=f"{CollatingSequences.NOCASE} (NVARCHAR)",
),
pytest.param(
CollatingSequences.RTRIM,
f"COLLATE {CollatingSequences.RTRIM}",
"NVARCHAR",
id=f"{CollatingSequences.RTRIM} (NVARCHAR)",
),
pytest.param(
CollatingSequences.BINARY,
"",
"TEXT",
id=f"{CollatingSequences.BINARY} (TEXT)",
),
pytest.param(
CollatingSequences.NOCASE,
f"COLLATE {CollatingSequences.NOCASE}",
"TEXT",
id=f"{CollatingSequences.NOCASE} (TEXT)",
),
pytest.param(
CollatingSequences.RTRIM,
f"COLLATE {CollatingSequences.RTRIM}",
"TEXT",
id=f"{CollatingSequences.RTRIM} (TEXT)",
),
pytest.param(
CollatingSequences.BINARY,
"",
"VARCHAR",
id=f"{CollatingSequences.BINARY} (VARCHAR)",
),
pytest.param(
CollatingSequences.NOCASE,
f"COLLATE {CollatingSequences.NOCASE}",
"VARCHAR",
id=f"{CollatingSequences.NOCASE} (VARCHAR)",
),
pytest.param(
CollatingSequences.RTRIM,
f"COLLATE {CollatingSequences.RTRIM}",
"VARCHAR",
id=f"{CollatingSequences.RTRIM} (VARCHAR)",
),
],
)
def test_data_type_collation_sequence_is_applied_on_textual_data_types(
self,
collation: str,
resulting_column_collation: str,
column_type: str,
) -> None:
assert MySQLtoSQLite._data_type_collation_sequence(collation, column_type) == resulting_column_collation
def test_data_type_collation_sequence_is_not_applied_on_non_textual_data_types(self) -> None:
for column_type in (
"BIGINT",
"BINARY",
"BIT",
"BLOB",
"BOOLEAN",
"DATE",
"DATETIME",
"DATETIME",
"DECIMAL",
"DOUBLE",
"FLOAT",
"INTEGER",
"INTEGER",
"LONGBLOB",
"MEDIUMBLOB",
"MEDIUMINT",
"NUMERIC",
"REAL",
"SMALLINT",
"TIME",
"TINYBLOB",
"TINYINT",
"VARBINARY",
"YEAR",
):
for collation in (
CollatingSequences.BINARY,
CollatingSequences.NOCASE,
CollatingSequences.RTRIM,
):
assert MySQLtoSQLite._data_type_collation_sequence(collation, column_type) == ""
@pytest.mark.exceptions
@pytest.mark.usefixtures("mysql_instance")
class TestMySQLtoSQLiteSQLExceptions:
@pytest.mark.parametrize(
"quiet",
[
pytest.param(False, id="verbose"),
pytest.param(True, id="quiet"),
],
)
def test_create_table_server_lost_connection_error(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
mocker: MockerFixture,
caplog: LogCaptureFixture,
quiet: bool,
) -> None:
proc: MySQLtoSQLite = MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=mysql_credentials.user,
mysql_password=mysql_credentials.password,
mysql_database=mysql_credentials.database,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
quiet=quiet,
)
class FakeSQLiteCursor:
def executescript(self, *args, **kwargs) -> t.Any:
raise mysql.connector.Error(
msg="Error Code: 2013. Lost connection to MySQL server during query",
errno=errorcode.CR_SERVER_LOST,
)
class FakeSQLiteConnector:
def commit(self, *args, **kwargs) -> t.Any:
return True
mysql_inspect: Inspector = inspect(mysql_database.engine)
mysql_tables: t.List[str] = mysql_inspect.get_table_names()
mocker.patch.object(proc, "_sqlite_cur", FakeSQLiteCursor())
mocker.patch.object(proc._mysql, "reconnect", return_value=True)
mocker.patch.object(proc, "_sqlite", FakeSQLiteConnector())
caplog.set_level(logging.DEBUG)
with pytest.raises(mysql.connector.Error):
proc._create_table(choice(mysql_tables))
@pytest.mark.parametrize(
"quiet",
[
pytest.param(False, id="verbose"),
pytest.param(True, id="quiet"),
],
)
def test_create_table_unknown_mysql_connector_error(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
mocker: MockerFixture,
caplog: LogCaptureFixture,
quiet: bool,
) -> None:
proc: MySQLtoSQLite = MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=mysql_credentials.user,
mysql_password=mysql_credentials.password,
mysql_database=mysql_credentials.database,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
quiet=quiet,
)
class FakeSQLiteCursor:
def executescript(self, statement: t.Any) -> t.Any:
raise mysql.connector.Error(
msg="Error Code: 2000. Unknown MySQL error",
errno=errorcode.CR_UNKNOWN_ERROR,
)
mysql_inspect: Inspector = inspect(mysql_database.engine)
mysql_tables: t.List[str] = mysql_inspect.get_table_names()
mocker.patch.object(proc, "_sqlite_cur", FakeSQLiteCursor())
caplog.set_level(logging.DEBUG)
with pytest.raises(mysql.connector.Error):
proc._create_table(choice(mysql_tables))
@pytest.mark.parametrize(
"quiet",
[
pytest.param(False, id="verbose"),
pytest.param(True, id="quiet"),
],
)
def test_create_table_sqlite3_error(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
mocker: MockerFixture,
caplog: LogCaptureFixture,
quiet: bool,
) -> None:
proc: MySQLtoSQLite = MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=mysql_credentials.user,
mysql_password=mysql_credentials.password,
mysql_database=mysql_credentials.database,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
quiet=quiet,
)
class FakeSQLiteCursor:
def executescript(self, *args, **kwargs) -> t.Any:
raise sqlite3.Error("Unknown SQLite error")
mysql_inspect: Inspector = inspect(mysql_database.engine)
mysql_tables: t.List[str] = mysql_inspect.get_table_names()
mocker.patch.object(proc, "_sqlite_cur", FakeSQLiteCursor())
caplog.set_level(logging.DEBUG)
with pytest.raises(sqlite3.Error):
proc._create_table(choice(mysql_tables))
@pytest.mark.parametrize(
"exception, quiet",
[
pytest.param(
mysql.connector.Error(
msg="Error Code: 2013. Lost connection to MySQL server during query",
errno=errorcode.CR_SERVER_LOST,
),
False,
id="errorcode.CR_SERVER_LOST verbose",
),
pytest.param(
mysql.connector.Error(
msg="Error Code: 2013. Lost connection to MySQL server during query",
errno=errorcode.CR_SERVER_LOST,
),
True,
id="errorcode.CR_SERVER_LOST quiet",
),
pytest.param(
mysql.connector.Error(
msg="Error Code: 2000. Unknown MySQL error",
errno=errorcode.CR_UNKNOWN_ERROR,
),
False,
id="errorcode.CR_UNKNOWN_ERROR verbose",
),
pytest.param(
mysql.connector.Error(
msg="Error Code: 2000. Unknown MySQL error",
errno=errorcode.CR_UNKNOWN_ERROR,
),
True,
id="errorcode.CR_UNKNOWN_ERROR quiet",
),
pytest.param(sqlite3.Error("Unknown SQLite error"), False, id="sqlite3.Error verbose"),
pytest.param(sqlite3.Error("Unknown SQLite error"), True, id="sqlite3.Error quiet"),
],
)
def test_transfer_table_data_exceptions(
self,
sqlite_database: "os.PathLike[t.Any]",
mysql_database: Database,
mysql_credentials: MySQLCredentials,
mocker: MockerFixture,
caplog: LogCaptureFixture,
exception: Exception,
quiet: bool,
) -> None:
proc: MySQLtoSQLite = MySQLtoSQLite( # type: ignore[call-arg]
sqlite_file=sqlite_database,
mysql_user=mysql_credentials.user,
mysql_password=mysql_credentials.password,
mysql_database=mysql_credentials.database,
mysql_host=mysql_credentials.host,
mysql_port=mysql_credentials.port,
quiet=quiet,
)
class FakeMySQLCursor:
def fetchall(self) -> t.Any:
raise exception
def fetchmany(self, size: int = 1) -> t.Any:
raise exception
mysql_inspect: Inspector = inspect(mysql_database.engine)
mysql_tables: t.List[str] = mysql_inspect.get_table_names()
table_name: str = choice(mysql_tables)
columns: t.List[str] = [column["name"] for column in mysql_inspect.get_columns(table_name)]
sql: str = 'INSERT OR IGNORE INTO "{table}" ({fields}) VALUES ({placeholders})'.format(
table=table_name,
fields=('"{}", ' * len(columns)).rstrip(" ,").format(*columns),
placeholders=("?, " * len(columns)).rstrip(" ,"),
)
mocker.patch.object(proc, "_mysql_cur", FakeMySQLCursor())
with pytest.raises((mysql.connector.Error, sqlite3.Error)):
proc._transfer_table_data(table_name, sql)
0707010000003A000081A4000000000000000000000001682E58C10000144E000000000000000000000000000000000000003600000000mysql-to-sqlite3-2.4.1/tests/unit/test_click_utils.pyimport typing as t
import click
import pytest
from click.testing import CliRunner
from pytest_mock import MockFixture
from mysql_to_sqlite3.click_utils import OptionEatAll, prompt_password, validate_positive_integer
class TestOptionEatAll:
def test_init_with_invalid_nargs(self) -> None:
"""Test OptionEatAll initialization with invalid nargs."""
with pytest.raises(ValueError) as excinfo:
OptionEatAll("--test", nargs=1)
assert "nargs, if set, must be -1 not 1" in str(excinfo.value)
def test_init_with_valid_nargs(self) -> None:
"""Test OptionEatAll initialization with valid nargs and behavior."""
@click.command()
@click.option("--test", cls=OptionEatAll, nargs=-1, help="Test option")
def cli(test: t.Optional[t.Tuple[str, ...]] = None) -> None:
# This just verifies that the option works when nargs=-1
assert test is not None
click.echo(f"Success: {len(test)} values")
runner = CliRunner()
result = runner.invoke(cli, ["--test", "value1", "value2", "value3"])
assert result.exit_code == 0
assert "Success:" in result.output
assert "values" in result.output
def test_add_to_parser(self) -> None:
"""Test add_to_parser method."""
@click.command()
@click.option("--test", cls=OptionEatAll, help="Test option")
def cli(test: t.Optional[t.Tuple[str, ...]] = None) -> None:
click.echo(f"Test: {test}")
runner = CliRunner()
result = runner.invoke(cli, ["--test", "value1", "value2", "value3"])
assert result.exit_code == 0
assert "Test: ('value1', 'value2', 'value3')" in result.output
def test_add_to_parser_with_other_options(self) -> None:
"""Test add_to_parser method with other options."""
@click.command()
@click.option("--test", cls=OptionEatAll, help="Test option")
@click.option("--other", help="Other option")
def cli(test: t.Optional[t.Tuple[str, ...]] = None, other: t.Optional[str] = None) -> None:
click.echo(f"Test: {test}, Other: {other}")
runner = CliRunner()
result = runner.invoke(cli, ["--test", "value1", "value2", "--other", "value3"])
assert result.exit_code == 0
assert "Test: ('value1', 'value2'), Other: value3" in result.output
def test_add_to_parser_without_save_other_options(self) -> None:
"""Test add_to_parser method without saving other options."""
@click.command()
@click.option("--test", cls=OptionEatAll, save_other_options=False, help="Test option")
@click.option("--other", help="Other option")
def cli(test: t.Optional[t.Tuple[str, ...]] = None, other: t.Optional[str] = None) -> None:
click.echo(f"Test: {test}, Other: {other}")
runner = CliRunner()
result = runner.invoke(cli, ["--test", "value1", "value2", "--other", "value3"])
assert result.exit_code == 0
# All remaining args should be consumed by --test
assert "Test: ('value1', 'value2', '--other', 'value3'), Other: None" in result.output
class TestPromptPassword:
def test_prompt_password_with_password(self) -> None:
"""Test prompt_password with password already provided."""
ctx = click.Context(click.Command("test"))
ctx.params = {"mysql_password": "test_password"}
result = prompt_password(ctx, None, True)
assert result == "test_password"
def test_prompt_password_without_password(self, mocker: MockFixture) -> None:
"""Test prompt_password without password provided."""
ctx = click.Context(click.Command("test"))
ctx.params = {"mysql_password": None}
mocker.patch("click.prompt", return_value="prompted_password")
result = prompt_password(ctx, None, True)
assert result == "prompted_password"
def test_prompt_password_use_password_false(self) -> None:
"""Test prompt_password with use_password=False."""
ctx = click.Context(click.Command("test"))
ctx.params = {"mysql_password": "test_password"}
result = prompt_password(ctx, None, False)
assert result is None
class TestValidatePositiveInteger:
def test_validate_positive_integer_valid(self) -> None:
"""Test validate_positive_integer with valid values."""
ctx = click.Context(click.Command("test"))
assert validate_positive_integer(ctx, None, 0) == 0
assert validate_positive_integer(ctx, None, 1) == 1
assert validate_positive_integer(ctx, None, 100) == 100
def test_validate_positive_integer_invalid(self) -> None:
"""Test validate_positive_integer with invalid values."""
ctx = click.Context(click.Command("test"))
with pytest.raises(click.BadParameter) as excinfo:
validate_positive_integer(ctx, None, -1)
assert "Should be a positive integer or 0." in str(excinfo.value)
with pytest.raises(click.BadParameter) as excinfo:
validate_positive_integer(ctx, None, -100)
assert "Should be a positive integer or 0." in str(excinfo.value)
0707010000003B000081A4000000000000000000000001682E58C100001787000000000000000000000000000000000000003500000000mysql-to-sqlite3-2.4.1/tests/unit/test_debug_info.pyimport sys
import typing as t
from unittest.mock import MagicMock, patch
import pytest
from pytest_mock import MockFixture
from mysql_to_sqlite3.debug_info import _implementation, _mysql_version, info
class TestDebugInfo:
def test_implementation_cpython(self, mocker: MockFixture) -> None:
"""Test _implementation function with CPython."""
mocker.patch("platform.python_implementation", return_value="CPython")
mocker.patch("platform.python_version", return_value="3.8.10")
result = _implementation()
assert result == "CPython 3.8.10"
def test_implementation_pypy(self, mocker: MockFixture) -> None:
"""Test _implementation function with PyPy."""
mocker.patch("platform.python_implementation", return_value="PyPy")
# Create a mock for pypy_version_info
mock_version_info = MagicMock()
mock_version_info.major = 3
mock_version_info.minor = 7
mock_version_info.micro = 4
mock_version_info.releaselevel = "final"
# Need to use patch instead of mocker.patch for sys module attributes
with patch.object(sys, "pypy_version_info", mock_version_info, create=True):
result = _implementation()
assert result == "PyPy 3.7.4"
def test_implementation_pypy_non_final(self, mocker: MockFixture) -> None:
"""Test _implementation function with PyPy non-final release."""
mocker.patch("platform.python_implementation", return_value="PyPy")
# Create a mock for pypy_version_info
mock_version_info = MagicMock()
mock_version_info.major = 3
mock_version_info.minor = 7
mock_version_info.micro = 4
mock_version_info.releaselevel = "beta"
# Need to use patch instead of mocker.patch for sys module attributes
with patch.object(sys, "pypy_version_info", mock_version_info, create=True):
result = _implementation()
assert result == "PyPy 3.7.4beta"
def test_implementation_jython(self, mocker: MockFixture) -> None:
"""Test _implementation function with Jython."""
mocker.patch("platform.python_implementation", return_value="Jython")
mocker.patch("platform.python_version", return_value="2.7.2")
result = _implementation()
assert result == "Jython 2.7.2"
def test_implementation_ironpython(self, mocker: MockFixture) -> None:
"""Test _implementation function with IronPython."""
mocker.patch("platform.python_implementation", return_value="IronPython")
mocker.patch("platform.python_version", return_value="2.7.9")
result = _implementation()
assert result == "IronPython 2.7.9"
def test_implementation_unknown(self, mocker: MockFixture) -> None:
"""Test _implementation function with unknown implementation."""
mocker.patch("platform.python_implementation", return_value="UnknownPython")
result = _implementation()
assert result == "UnknownPython Unknown"
def test_mysql_version_success(self, mocker: MockFixture) -> None:
"""Test _mysql_version function when mysql client is available."""
mocker.patch("mysql_to_sqlite3.debug_info.which", return_value="/usr/bin/mysql")
mocker.patch(
"mysql_to_sqlite3.debug_info.check_output",
return_value=b"mysql Ver 8.0.26 for Linux on x86_64",
)
result = _mysql_version()
assert result == "mysql Ver 8.0.26 for Linux on x86_64"
def test_mysql_version_bytes_decode_error(self, mocker: MockFixture) -> None:
"""Test _mysql_version function when bytes decoding fails."""
mocker.patch("mysql_to_sqlite3.debug_info.which", return_value="/usr/bin/mysql")
mock_output = MagicMock()
mock_output.decode.side_effect = UnicodeDecodeError("utf-8", b"", 0, 1, "invalid")
mocker.patch(
"mysql_to_sqlite3.debug_info.check_output",
return_value=mock_output,
)
result = _mysql_version()
assert isinstance(result, str)
def test_mysql_version_exception(self, mocker: MockFixture) -> None:
"""Test _mysql_version function when an exception occurs."""
mocker.patch("mysql_to_sqlite3.debug_info.which", return_value="/usr/bin/mysql")
mocker.patch(
"mysql_to_sqlite3.debug_info.check_output",
side_effect=Exception("Command failed"),
)
result = _mysql_version()
assert result == "MySQL client not found on the system"
def test_mysql_version_not_found(self, mocker: MockFixture) -> None:
"""Test _mysql_version function when mysql client is not found."""
mocker.patch("mysql_to_sqlite3.debug_info.which", return_value=None)
result = _mysql_version()
assert result == "MySQL client not found on the system"
def test_info_success(self, mocker: MockFixture) -> None:
"""Test info function."""
mocker.patch("platform.system", return_value="Linux")
mocker.patch("platform.release", return_value="5.4.0-80-generic")
mocker.patch("mysql_to_sqlite3.debug_info._implementation", return_value="CPython 3.8.10")
mocker.patch("mysql_to_sqlite3.debug_info._mysql_version", return_value="mysql Ver 8.0.26 for Linux on x86_64")
result = info()
assert isinstance(result, list)
assert len(result) > 0
assert result[2] == ["Operating System", "Linux 5.4.0-80-generic"]
assert result[3] == ["Python", "CPython 3.8.10"]
assert result[4] == ["MySQL", "mysql Ver 8.0.26 for Linux on x86_64"]
def test_info_platform_error(self, mocker: MockFixture) -> None:
"""Test info function when platform.system raises IOError."""
mocker.patch("platform.system", side_effect=IOError("Platform error"))
result = info()
assert isinstance(result, list)
assert len(result) > 0
assert result[2] == ["Operating System", "Unknown"]
0707010000003C000081A4000000000000000000000001682E58C100001F1B000000000000000000000000000000000000003600000000mysql-to-sqlite3-2.4.1/tests/unit/test_mysql_utils.py"""Unit tests for the mysql_utils module."""
import typing as t
from unittest import mock
import pytest
from mysql.connector import CharacterSet
from mysql_to_sqlite3.mysql_utils import (
CHARSET_INTRODUCERS,
CharSet,
mysql_supported_character_sets,
)
class TestMySQLUtils:
"""Unit tests for the mysql_utils module."""
def test_charset_introducers(self) -> None:
"""Test that CHARSET_INTRODUCERS contains the expected values."""
assert isinstance(CHARSET_INTRODUCERS, tuple)
assert len(CHARSET_INTRODUCERS) > 0
assert all(isinstance(intro, str) for intro in CHARSET_INTRODUCERS)
assert all(intro.startswith("_") for intro in CHARSET_INTRODUCERS)
def test_charset_named_tuple(self) -> None:
"""Test the CharSet named tuple."""
charset = CharSet(id=1, charset="utf8", collation="utf8_general_ci")
assert charset.id == 1
assert charset.charset == "utf8"
assert charset.collation == "utf8_general_ci"
def test_mysql_supported_character_sets_with_charset(self) -> None:
"""Test mysql_supported_character_sets with a specific charset."""
test_charset = "utf8mb4"
results = list(mysql_supported_character_sets(test_charset))
assert len(results) > 0
for result in results:
assert result.charset == test_charset
assert isinstance(result.id, int)
assert isinstance(result.collation, str)
@mock.patch("mysql_to_sqlite3.mysql_utils.MYSQL_CHARACTER_SETS", [(None, None), ("utf8", "utf8_general_ci", True)])
def test_mysql_supported_character_sets_with_charset_keyerror(self) -> None:
"""Test handling KeyError in mysql_supported_character_sets with charset."""
# Override the MYSQL_CHARACTER_SETS behavior to raise KeyError
with mock.patch("mysql_to_sqlite3.mysql_utils.MYSQL_CHARACTER_SETS") as mock_charset_sets:
mock_charset_sets.__getitem__.side_effect = KeyError("Test KeyError")
# This should not raise any exceptions
results = list(mysql_supported_character_sets("utf8"))
assert len(results) == 0
@mock.patch("mysql_to_sqlite3.mysql_utils.CharacterSet")
def test_mysql_supported_character_sets_no_charset(self, mock_charset_class) -> None:
"""Test mysql_supported_character_sets with no charset."""
# Create a custom mock class for CharacterSet that has get_supported method
mock_instance = mock.MagicMock()
mock_instance.get_supported.return_value = ["utf8"]
mock_charset_class.return_value = mock_instance
with mock.patch(
"mysql_to_sqlite3.mysql_utils.MYSQL_CHARACTER_SETS",
[
None, # This will be skipped due to None check
("utf8", "utf8_general_ci", True), # This will be processed
],
):
results = list(mysql_supported_character_sets())
mock_instance.get_supported.assert_called_once()
assert len(results) > 0
charset_results = [r.charset for r in results]
assert "utf8" in charset_results
@mock.patch("mysql_to_sqlite3.mysql_utils.CharacterSet")
def test_mysql_supported_character_sets_no_charset_keyerror(self, mock_charset_class) -> None:
"""Test handling KeyError in mysql_supported_character_sets without charset."""
# Setup mock to return specific values
mock_instance = mock.MagicMock()
mock_instance.get_supported.return_value = ["utf8"]
mock_charset_class.return_value = mock_instance
# Create a mock object to return either valid info or raise KeyError
mock_char_sets = mock.MagicMock()
mock_char_sets.__len__.return_value = 2
# Make the first item None and the second one raise KeyError when accessed
def getitem_side_effect(idx):
if idx == 0:
return (None, None, None)
else:
raise KeyError("Test KeyError")
mock_char_sets.__getitem__.side_effect = getitem_side_effect
# Patch with the mock object
with mock.patch("mysql_to_sqlite3.mysql_utils.MYSQL_CHARACTER_SETS", mock_char_sets):
# Should continue without raising exceptions despite KeyError
results = list(mysql_supported_character_sets())
assert len(results) == 0
# Verify the get_supported method was called
mock_instance.get_supported.assert_called_once()
def test_mysql_supported_character_sets_complete_coverage(self) -> None:
"""Test mysql_supported_character_sets to target specific edge cases for full coverage."""
# Test with a charset that doesn't match any entries
with mock.patch(
"mysql_to_sqlite3.mysql_utils.MYSQL_CHARACTER_SETS",
[("utf8", "utf8_general_ci", True), None, ("latin1", "latin1_swedish_ci", True)], # Test None handling
):
# Should return empty when charset doesn't match any entries
results = list(mysql_supported_character_sets("non_existent_charset"))
assert len(results) == 0
# Should process all valid charsets when no specific charset is requested
with mock.patch("mysql_to_sqlite3.mysql_utils.CharacterSet") as mock_charset_class:
mock_instance = mock.MagicMock()
mock_instance.get_supported.return_value = ["utf8", "latin1", "invalid_charset"]
mock_charset_class.return_value = mock_instance
# Test when no charset is specified - should process all entries
results = list(mysql_supported_character_sets())
# Should have entries for utf8 and latin1
assert len([r for r in results if r.charset in ["utf8", "latin1"]]) > 0
def test_mysql_supported_character_sets_with_specific_keyerror(self) -> None:
"""Test mysql_supported_character_sets with specific KeyError scenarios."""
# Test the specific KeyError scenario in the first branch (with charset specified)
with mock.patch(
"mysql_to_sqlite3.mysql_utils.MYSQL_CHARACTER_SETS",
[
None,
mock.MagicMock(side_effect=KeyError("Key error in info[0]")), # Trigger KeyError on info[0]
("latin1", "latin1_swedish_ci", True),
],
):
# This should not raise exceptions despite the KeyError
results = list(mysql_supported_character_sets("utf8"))
assert len(results) == 0
@mock.patch("mysql_to_sqlite3.mysql_utils.CharacterSet")
def test_mysql_supported_character_sets_with_specific_info_keyerror(self, mock_charset_class) -> None:
"""Test mysql_supported_character_sets with KeyError on info[1] access."""
mock_instance = mock.MagicMock()
mock_instance.get_supported.return_value = ["utf8"]
mock_charset_class.return_value = mock_instance
# Create a special mock that will raise KeyError on accessing index 1
class InfoMock:
def __getitem__(self, key):
if key == 0:
return "utf8"
elif key == 1:
raise KeyError("Test KeyError on info[1]")
return None
# Set up MYSQL_CHARACTER_SETS with our special mock
with mock.patch(
"mysql_to_sqlite3.mysql_utils.MYSQL_CHARACTER_SETS",
[
None,
InfoMock(), # Will raise KeyError on info[1]
],
):
# This should not raise exceptions despite the KeyError
results = list(mysql_supported_character_sets())
assert len(results) == 0
# Now test with a specific charset to cover both branches
results = list(mysql_supported_character_sets("utf8"))
assert len(results) == 0
0707010000003D000081A4000000000000000000000001682E58C1000015A6000000000000000000000000000000000000003700000000mysql-to-sqlite3-2.4.1/tests/unit/test_sqlite_utils.pyfrom datetime import date, timedelta
from decimal import Decimal
import pytest
from mysql_to_sqlite3.sqlite_utils import (
CollatingSequences,
Integer_Types,
adapt_decimal,
adapt_timedelta,
convert_date,
convert_decimal,
convert_timedelta,
encode_data_for_sqlite,
)
class TestSQLiteUtils:
def test_adapt_decimal(self) -> None:
"""Test adapt_decimal function."""
assert adapt_decimal(Decimal("123.45")) == "123.45"
assert adapt_decimal(Decimal("0")) == "0"
assert adapt_decimal(Decimal("-123.45")) == "-123.45"
assert adapt_decimal(Decimal("123456789.123456789")) == "123456789.123456789"
def test_convert_decimal(self) -> None:
"""Test convert_decimal function."""
assert convert_decimal("123.45") == Decimal("123.45")
assert convert_decimal("0") == Decimal("0")
assert convert_decimal("-123.45") == Decimal("-123.45")
assert convert_decimal("123456789.123456789") == Decimal("123456789.123456789")
def test_adapt_timedelta(self) -> None:
"""Test adapt_timedelta function."""
assert adapt_timedelta(timedelta(hours=1, minutes=30, seconds=45)) == "01:30:45"
assert adapt_timedelta(timedelta(hours=0, minutes=0, seconds=0)) == "00:00:00"
assert adapt_timedelta(timedelta(hours=100, minutes=0, seconds=0)) == "100:00:00"
assert adapt_timedelta(timedelta(hours=0, minutes=90, seconds=0)) == "01:30:00"
assert adapt_timedelta(timedelta(hours=0, minutes=0, seconds=90)) == "00:01:30"
def test_convert_timedelta(self) -> None:
"""Test convert_timedelta function."""
assert convert_timedelta("01:30:45") == timedelta(hours=1, minutes=30, seconds=45)
assert convert_timedelta("00:00:00") == timedelta(hours=0, minutes=0, seconds=0)
assert convert_timedelta("100:00:00") == timedelta(hours=100, minutes=0, seconds=0)
assert convert_timedelta("01:30:00") == timedelta(hours=1, minutes=30, seconds=0)
assert convert_timedelta("00:01:30") == timedelta(hours=0, minutes=1, seconds=30)
def test_encode_data_for_sqlite_string(self) -> None:
"""Test encode_data_for_sqlite with string."""
assert encode_data_for_sqlite("test") == "test"
def test_encode_data_for_sqlite_bytes_success(self) -> None:
"""Test encode_data_for_sqlite with bytes that can be decoded."""
assert encode_data_for_sqlite(b"test") == "test"
def test_encode_data_for_sqlite_bytes_failure(self) -> None:
"""Test encode_data_for_sqlite with bytes that cannot be decoded."""
# Create invalid UTF-8 bytes
invalid_bytes = b"\xff\xfe\xfd"
result = encode_data_for_sqlite(invalid_bytes)
# Should return a sqlite3.Binary object or something that behaves similarly
# Check if it's either a Binary object or at least contains the original bytes
if hasattr(result, "adapt"):
assert result.adapt() == invalid_bytes
else:
# If it's a memoryview or other type, verify it contains our original bytes
assert bytes(result) == invalid_bytes
def test_encode_data_for_sqlite_non_bytes(self) -> None:
"""Test encode_data_for_sqlite with non-bytes object that has no decode method."""
result = encode_data_for_sqlite(123)
# In our implementation, the function should either:
# 1. Return a sqlite3.Binary object, or
# 2. Return the original value if it can't be converted to Binary
# Either way, the result should work with SQLite
if hasattr(result, "adapt"):
assert result.adapt() == 123
else:
assert result == 123
def test_convert_date_valid_string(self) -> None:
"""Test convert_date with valid string."""
assert convert_date("2021-01-01") == date(2021, 1, 1)
assert convert_date("2021/01/01") == date(2021, 1, 1)
assert convert_date("Jan 1, 2021") == date(2021, 1, 1)
def test_convert_date_valid_bytes(self) -> None:
"""Test convert_date with valid bytes."""
assert convert_date(b"2021-01-01") == date(2021, 1, 1)
assert convert_date(b"2021/01/01") == date(2021, 1, 1)
assert convert_date(b"Jan 1, 2021") == date(2021, 1, 1)
def test_convert_date_invalid(self) -> None:
"""Test convert_date with invalid date string."""
with pytest.raises(ValueError) as excinfo:
convert_date("not a date")
assert "DATE field contains" in str(excinfo.value)
def test_collating_sequences(self) -> None:
"""Test CollatingSequences class."""
assert CollatingSequences.BINARY == "BINARY"
assert CollatingSequences.NOCASE == "NOCASE"
assert CollatingSequences.RTRIM == "RTRIM"
def test_integer_types(self) -> None:
"""Test Integer_Types set."""
assert "INTEGER" in Integer_Types
assert "INT" in Integer_Types
assert "BIGINT" in Integer_Types
assert "SMALLINT" in Integer_Types
assert "TINYINT" in Integer_Types
assert "MEDIUMINT" in Integer_Types
assert "NUMERIC" in Integer_Types
# Check that unsigned variants are included
assert "INTEGER UNSIGNED" in Integer_Types
assert "INT UNSIGNED" in Integer_Types
assert "BIGINT UNSIGNED" in Integer_Types
assert "SMALLINT UNSIGNED" in Integer_Types
assert "TINYINT UNSIGNED" in Integer_Types
assert "MEDIUMINT UNSIGNED" in Integer_Types
0707010000003E000081A4000000000000000000000001682E58C100002992000000000000000000000000000000000000003600000000mysql-to-sqlite3-2.4.1/tests/unit/test_transporter.pyimport sqlite3
from unittest.mock import MagicMock, patch
import pytest
from mysql_to_sqlite3.transporter import MySQLtoSQLite
class TestMySQLtoSQLiteTransporter:
def test_decode_column_type_with_string(self) -> None:
"""Test _decode_column_type with string input."""
assert MySQLtoSQLite._decode_column_type("VARCHAR") == "VARCHAR"
assert MySQLtoSQLite._decode_column_type("INTEGER") == "INTEGER"
assert MySQLtoSQLite._decode_column_type("TEXT") == "TEXT"
def test_decode_column_type_with_bytes(self) -> None:
"""Test _decode_column_type with bytes input."""
assert MySQLtoSQLite._decode_column_type(b"VARCHAR") == "VARCHAR"
assert MySQLtoSQLite._decode_column_type(b"INTEGER") == "INTEGER"
assert MySQLtoSQLite._decode_column_type(b"TEXT") == "TEXT"
def test_decode_column_type_with_bytes_decode_error(self) -> None:
"""Test _decode_column_type with bytes that fail to decode."""
# Create a mock bytes object that raises UnicodeDecodeError when decode is called
mock_bytes = MagicMock(spec=bytes)
mock_bytes.decode.side_effect = UnicodeDecodeError("utf-8", b"", 0, 1, "invalid")
# Patch the isinstance function to return True for our mock_bytes
with patch(
"mysql_to_sqlite3.transporter.isinstance",
lambda obj, cls: True if obj is mock_bytes and cls is bytes else isinstance(obj, cls),
):
result = MySQLtoSQLite._decode_column_type(mock_bytes)
assert isinstance(result, str)
# The string representation of the mock should be in the result
assert str(mock_bytes) in result
def test_decode_column_type_with_non_string_non_bytes(self) -> None:
"""Test _decode_column_type with input that is neither string nor bytes."""
assert MySQLtoSQLite._decode_column_type(123) == "123"
assert MySQLtoSQLite._decode_column_type(None) == "None"
assert MySQLtoSQLite._decode_column_type(True) == "True"
@patch("sqlite3.connect")
def test_check_sqlite_json1_extension_enabled_success(self, mock_connect: MagicMock) -> None:
"""Test _check_sqlite_json1_extension_enabled when JSON1 is enabled."""
# Setup mock cursor
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = [("ENABLE_JSON1",), ("ENABLE_FTS5",)]
# Setup mock connection
mock_connection = MagicMock()
mock_connection.cursor.return_value = mock_cursor
mock_connect.return_value = mock_connection
# Create a minimal instance with just what we need for the test
with patch.object(MySQLtoSQLite, "__init__", return_value=None):
instance = MySQLtoSQLite()
instance._sqlite_cur = mock_cursor
# Test the method
result = instance._check_sqlite_json1_extension_enabled()
assert result is True
mock_cursor.execute.assert_called_with("PRAGMA compile_options")
@patch("sqlite3.connect")
def test_check_sqlite_json1_extension_disabled(self, mock_connect: MagicMock) -> None:
"""Test _check_sqlite_json1_extension_enabled when JSON1 is not enabled."""
# Setup mock cursor
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = [("ENABLE_FTS5",), ("ENABLE_RTREE",)]
# Setup mock connection
mock_connection = MagicMock()
mock_connection.cursor.return_value = mock_cursor
mock_connect.return_value = mock_connection
# Create a minimal instance with just what we need for the test
with patch.object(MySQLtoSQLite, "__init__", return_value=None):
instance = MySQLtoSQLite()
instance._sqlite_cur = mock_cursor
# Test the method
result = instance._check_sqlite_json1_extension_enabled()
assert result is False
mock_cursor.execute.assert_called_with("PRAGMA compile_options")
@patch("sqlite3.connect")
def test_check_sqlite_json1_extension_error(self, mock_connect: MagicMock) -> None:
"""Test _check_sqlite_json1_extension_enabled when an error occurs."""
# Setup mock cursor
mock_cursor = MagicMock()
mock_cursor.execute.side_effect = sqlite3.Error("Test error")
# Setup mock connection
mock_connection = MagicMock()
mock_connection.cursor.return_value = mock_cursor
mock_connect.return_value = mock_connection
# Create a minimal instance with just what we need for the test
with patch.object(MySQLtoSQLite, "__init__", return_value=None):
instance = MySQLtoSQLite()
instance._sqlite_cur = mock_cursor
# Test the method
result = instance._check_sqlite_json1_extension_enabled()
assert result is False
mock_cursor.execute.assert_called_with("PRAGMA compile_options")
@patch("mysql.connector.connect")
@patch("sqlite3.connect")
def test_transfer_exception_handling(self, mock_sqlite_connect: MagicMock, mock_mysql_connect: MagicMock) -> None:
"""Test transfer method exception handling."""
# Setup mock SQLite cursor
mock_sqlite_cursor = MagicMock()
# Setup mock SQLite connection
mock_sqlite_connection = MagicMock()
mock_sqlite_connection.cursor.return_value = mock_sqlite_cursor
mock_sqlite_connect.return_value = mock_sqlite_connection
# Setup mock MySQL cursor
mock_mysql_cursor = MagicMock()
mock_mysql_cursor.fetchall.return_value = [(b"table1",)]
# Setup mock MySQL connection
mock_mysql_connection = MagicMock()
mock_mysql_connection.cursor.return_value = mock_mysql_cursor
mock_mysql_connect.return_value = mock_mysql_connection
# Create a minimal instance with just what we need for the test
with patch.object(MySQLtoSQLite, "__init__", return_value=None):
instance = MySQLtoSQLite()
instance._mysql_tables = []
instance._exclude_mysql_tables = []
instance._mysql_cur = mock_mysql_cursor
instance._sqlite_cur = mock_sqlite_cursor
instance._without_data = False
instance._without_tables = False
instance._vacuum = False
instance._logger = MagicMock()
# Mock the _create_table method to raise an exception
instance._create_table = MagicMock(side_effect=Exception("Test exception"))
# Test that the exception is properly propagated
with pytest.raises(Exception) as excinfo:
instance.transfer()
assert "Test exception" in str(excinfo.value)
# Verify that foreign keys are re-enabled in the finally block
mock_sqlite_cursor.execute.assert_called_with("PRAGMA foreign_keys=ON")
def test_constructor_missing_mysql_database(self) -> None:
"""Test constructor raises ValueError if mysql_database is missing."""
from mysql_to_sqlite3.transporter import MySQLtoSQLite
with pytest.raises(ValueError, match="Please provide a MySQL database"):
MySQLtoSQLite(mysql_user="user", sqlite_file="file.db")
def test_constructor_missing_mysql_user(self) -> None:
"""Test constructor raises ValueError if mysql_user is missing."""
from mysql_to_sqlite3.transporter import MySQLtoSQLite
with pytest.raises(ValueError, match="Please provide a MySQL user"):
MySQLtoSQLite(mysql_database="db", sqlite_file="file.db")
def test_constructor_missing_sqlite_file(self) -> None:
"""Test constructor raises ValueError if sqlite_file is missing."""
from mysql_to_sqlite3.transporter import MySQLtoSQLite
with pytest.raises(ValueError, match="Please provide an SQLite file"):
MySQLtoSQLite(mysql_database="db", mysql_user="user")
def test_constructor_mutually_exclusive_tables(self) -> None:
"""Test constructor raises ValueError if both mysql_tables and exclude_mysql_tables are provided."""
from mysql_to_sqlite3.transporter import MySQLtoSQLite
with pytest.raises(ValueError, match="mutually exclusive"):
MySQLtoSQLite(
mysql_database="db",
mysql_user="user",
sqlite_file="file.db",
mysql_tables=["a"],
exclude_mysql_tables=["b"],
)
def test_constructor_without_tables_and_data(self) -> None:
"""Test constructor raises ValueError if both without_tables and without_data are True."""
from mysql_to_sqlite3.transporter import MySQLtoSQLite
with pytest.raises(ValueError, match="Unable to continue without transferring data or creating tables!"):
MySQLtoSQLite(
mysql_database="db", mysql_user="user", sqlite_file="file.db", without_tables=True, without_data=True
)
def test_translate_default_from_mysql_to_sqlite_none(self) -> None:
"""Test _translate_default_from_mysql_to_sqlite with None default."""
assert MySQLtoSQLite._translate_default_from_mysql_to_sqlite(None) == ""
def test_translate_default_from_mysql_to_sqlite_bool(self) -> None:
"""Test _translate_default_from_mysql_to_sqlite with boolean default."""
assert MySQLtoSQLite._translate_default_from_mysql_to_sqlite(True, column_type="BOOLEAN") in (
"DEFAULT(TRUE)",
"DEFAULT '1'",
)
assert MySQLtoSQLite._translate_default_from_mysql_to_sqlite(False, column_type="BOOLEAN") in (
"DEFAULT(FALSE)",
"DEFAULT '0'",
)
def test_translate_default_from_mysql_to_sqlite_str(self) -> None:
"""Test _translate_default_from_mysql_to_sqlite with string default."""
assert MySQLtoSQLite._translate_default_from_mysql_to_sqlite("test") == "DEFAULT 'test'"
def test_translate_default_from_mysql_to_sqlite_current_timestamp(self) -> None:
"""Test _translate_default_from_mysql_to_sqlite with CURRENT_TIMESTAMP."""
assert (
MySQLtoSQLite._translate_default_from_mysql_to_sqlite("CURRENT_TIMESTAMP", column_extra="DEFAULT_GENERATED")
== "DEFAULT CURRENT_TIMESTAMP"
)
def test_translate_default_from_mysql_to_sqlite_bytes(self) -> None:
"""Test _translate_default_from_mysql_to_sqlite with bytes default."""
result = MySQLtoSQLite._translate_default_from_mysql_to_sqlite(b"abc", column_type="BLOB")
assert result.startswith("DEFAULT x'")
0707010000003F000081A4000000000000000000000001682E58C100000841000000000000000000000000000000000000001F00000000mysql-to-sqlite3-2.4.1/tox.ini[tox]
isolated_build = true
envlist =
python3.9,
python3.10,
python3.11,
python3.12,
python3.13,
black,
flake8,
linters
skip_missing_interpreters = true
[gh-actions]
python =
3.9: python3.9
3.10: python3.10
3.11: python3.11
3.12: python3.12
3.13: python3.13
[testenv]
passenv =
LANG
LEGACY_DB
deps =
-rrequirements_dev.txt
commands = pytest -v --cov=src/mysql_to_sqlite3 --cov-report=xml
[testenv:black]
basepython = python3
skip_install = true
deps =
black
commands = black src/mysql_to_sqlite3 tests/
[testenv:isort]
basepython = python3
skip_install = true
deps =
isort
commands =
isort --check-only --diff .
[testenv:flake8]
basepython = python3
skip_install = true
deps =
flake8
flake8-colors
flake8-docstrings
flake8-import-order
flake8-typing-imports
pep8-naming
commands =
flake8 src/mysql_to_sqlite3
[testenv:pylint]
basepython = python3
skip_install = true
deps =
pylint
-rrequirements_dev.txt
disable = C0209,C0301,C0411,R,W0107,W0622
commands =
pylint --rcfile=tox.ini src/mysql_to_sqlite3
[testenv:bandit]
basepython = python3
skip_install = true
deps =
bandit
commands =
bandit -r src/mysql_to_sqlite3 -c .bandit.yml
[testenv:mypy]
basepython = python3
skip_install = true
deps =
mypy>=1.3.0
-rrequirements_dev.txt
commands =
mypy src/mysql_to_sqlite3
[testenv:linters]
basepython = python3
skip_install = true
deps =
{[testenv:black]deps}
{[testenv:isort]deps}
{[testenv:flake8]deps}
{[testenv:pylint]deps}
{[testenv:bandit]deps}
{[testenv:mypy]deps}
commands =
{[testenv:black]commands}
{[testenv:isort]commands}
{[testenv:flake8]commands}
{[testenv:pylint]commands}
{[testenv:bandit]commands}
{[testenv:mypy]commands}
[flake8]
ignore = I100,I201,I202,D203,D401,W503,E203,F401,F403,C901,E501
exclude =
*__init__.py
*__version__.py
.tox
max-complexity = 10
max-line-length = 88
import-order-style = pycharm
application-import-names = flake8
[pylint]
disable = C0209,C0301,C0411,R,W0107,W062207070100000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000B00000000TRAILER!!!579 blocks
