标签: sqlalchemy-utils

我对 Python 很陌生，对使用 postgresql 也很陌生，所以如果这是基本的东西（我 - 到目前为止 - 未能实现），请原谅我。我正在尝试编写一个 python 代码：

• 创建一个新的数据库 ( testdb)
• 将 csv 文件读入 Pandas 数据帧
• 从 Pandas 数据框中创建并填充数据库中的新表。

到目前为止，我有 3 个不同的文件：a）一个.ini-File，我在其中存储创建新数据库所需的数据库信息，b）一个.csv-File（来自此处，命名为100_recs.csv）和 c）我的 python 代码。

数据库.ini :

[postgresql]
host=localhost
user=postgres
password=creator
port=5432

db_creator.py :

from config import config
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database
import pandas as pd

# useful info for psycopg2:
# /sf/ask/2413884651/


class MyDB(object):
    def __init__(self):
        self.params = config()

    def …

我正在尝试创建一个使用 UUID 作为主键的用户模型：

from src.db import db # SQLAlchemy instance

import sqlalchemy_utils

import uuid


class User(db.Model):
    __tablename__ = 'user'

    id = db.Column(sqlalchemy_utils.UUIDType(binary=True), primary_key=True, nullable=False)

但是当我生成迁移时，我收到：

File "/home/pc/Downloads/project/auth/venv/lib/python3.6/site-packages/alembic/runtime/environment.py", line 836, in run_migrations
    self.get_context().run_migrations(**kw)
  File "/home/pc/Downloads/project/auth/venv/lib/python3.6/site-packages/alembic/runtime/migration.py", line 330, in run_migrations
step.migration_fn(**kw)
  File "/home/pc/Downloads/project/auth/migrations/versions/efae4166f832_.py", line 22, in upgrade
    sa.Column('id', sqlalchemy_utils.types.uuid.UUIDType(length=16), nullable=False),
NameError: name 'sqlalchemy_utils' is not defined`

我不得不尽量明确地告诉我使用像模块，这和使用一个“内部”的实现，SQLAlchemy的

数据：如果我manualy导入sqlalchemy_utils的/migrations/version/efae4166f832_.py并卸下automaticaly产生的长度sa.Column('id', sqlalchemy_utils.types.uuid.UUIDType(length=16), nullable=False)它的工作原理罚款

我使用generate.py脚本生成迁移：

from src import create_app

from src.db import db

from …

列类型的SQLAlchemy-Utils 文档EncryptedType有一个如下所示的示例：

secret_key = 'secretkey1234'
# setup
engine = create_engine('sqlite:///:memory:')
connection = engine.connect()
Base = declarative_base()


class User(Base):
    __tablename__ = "user"
    id = sa.Column(sa.Integer, primary_key=True)
    username = sa.Column(EncryptedType(sa.Unicode,
                                       secret_key,
                                       AesEngine,
                                       'pkcs5'))

但是如果我在定义User类之前不知道密钥是什么怎么办？例如，如果我想提示用户输入密钥怎么办？

我一直在尝试使用 sqlalchemy 的 bulk_insert_mappings。我知道我可以创建会话并连接到数据库。我已经初始化了我的引擎，但我似乎无法从表中获得我需要的映射器。

from sqlalchemy import create_engine
from sqlalchemy.orm.session import sessionmaker,Session
from sqlalchemy_utils import get_mapper

engine = create_engine('mysql+pymysql://{}:{}@IP:PORT/'.format(USER,PW)) # removed my config here
connection = engine.connect()
m = MetaData(bind=engine,schema='test')
m.reflect()

Session = sessionmaker(bind=engine)
s = Session()
s.bulk_insert_mappings(get_mapper(m.tables['test.test']), pd.DataFrame({'a':['a','b','c']}).to_dict(orient="records"))
s.commit()
s.close()

我最近在这个问题上发现了一堆相关的问题

SQLAlchemy 从 Table 对象中获取 Mapper 对象（来自 Metadata 或 Session 或其他）

但 sqlalchemy_utils.get_mapper 提出：

“ValueError: 无法获得表 'test' 的映射器。”

sqlalchemy.orm.mapperlib._mapper_registry似乎是空的。也许是因为我没有将它绑定到我的引擎上。但不知道该怎么做。

PS：test是一个非常简单的TEXT类型的单列表

这是 m.tables['test.test'] 的输出

Table('test', MetaData(bind=Engine(mysql+pymysql://USER:***@IP:PORT/)), Column('a', TEXT(), table=<test>), schema='test')