标签: peewee

我有以下脚本:

from peewee import *

db = MySQLDatabase('database', user='root')

class BaseModel(Model):
    class Meta:
        database = db

class Locations(BaseModel):
    location_id = PrimaryKeyField()
    location_name = CharField()

class Units(BaseModel):
    unit_id = PrimaryKeyField()
    unit_num = IntegerField()
    location_id = ForeignKeyField(Locations, related_name='units')

db.connect()

for location in Locations.select():
    for pod_num in range (1, 9):
        unit = Units.create(unit_num=pod_num, location_id=location.location_id)

表位置有几行,表单位为空.当我尝试启动它时,我会一直异常:

(1054, "Unknown column 'location_id_id' in 'field list'")

我究竟做错了什么？

以下是用于创建表的SQL脚本的一部分:

CREATE  TABLE IF NOT EXISTS `database`.`units` (
  `unit_id` INT UNSIGNED NOT NULL AUTO_INCREMENT ,
  `unit_num` TINYINT UNSIGNED NOT NULL …

嗨,我正在使用Flask Peewee并尝试更新merchant_details模型,但它无法正常工作.以下是我得到的错误:

AttributeError:'SelectQuery'对象没有属性'update'

mdetails = merchant_details.filter(merchant_details.merchant_id==session['userid']).update(
         merchant_name=request.form['merchantname'],
          first_name=request.form['firstname'],
          last_name=request.form['lastname'],
        )

请帮忙!

这有效:

class MyModel(peewee.Model):
    my_field = peewee.IntegerField(null=False, default=0)

    class Meta(object):
        database = db
        db_table = 'MyTable'

这不起作用:

class MyModel(peewee.Model):

    class Meta(object):
        database = db
        db_table = 'MyTable'

setattr(MyModel, 'my_field', peewee.IntegerField(null=False, default=0))

我想这是由于某些元类魔法所做的peewee.Model.实际上,我无法在模型中看到它_meta.fields.

什么是动态定义字段的好方法？

我从Python的peewee模块得到一个奇怪的错误,我无法解决,任何想法？我基本上希望拥有包含多个公司的"批次".我正在为每个批处理创建一个批处理实例,并将其中的所有公司分配给该批处理的行ID.

追溯

Traceback (most recent call last):
  File "app.py", line 16, in <module>
    import models
  File "/Users/wyssuser/Desktop/dscraper/models.py", line 10, in <module>
    class Batch(Model):
  File "/Library/Python/2.7/site-packages/peewee.py", line 3647, in __new__
    cls._meta.prepared()
  File "/Library/Python/2.7/site-packages/peewee.py", line 3497, in prepared
    field = self.fields[item.lstrip('-')]
KeyError: 'i'

models.py

from datetime import datetime

from flask.ext.bcrypt import generate_password_hash
from flask.ext.login import UserMixin

from peewee import *

DATABASE = SqliteDatabase('engineering.db')

class Batch(Model):
  initial_contact_date = DateTimeField(formats="%m-%d-%Y")

  class Meta:
    database = DATABASE
    order_by = ('initial_contact_date')

  @classmethod
  def create_batch(cls, initial_contact_date):
    try:
      with …

我用烧瓶和小便.有时peewee会抛出这个错误

MySQL server has gone away (error(32, 'Broken pipe'))

Peewee数据库连接

db = PooledMySQLDatabase(database,**{
            "passwd": password, "user": user,
            "max_connections":None,"stale_timeout":None,
            "threadlocals" : True
        })

@app.before_request
def before_request():
    db.connect()

@app.teardown_request
def teardown_request(exception):
    db.close()

在mysql错误"MySQL服务器已经消失(错误(32,'Broken pipe'))"之后,选择查询工作没有问题,但插入,更新,删除查询不起作用.

在插入,更新,删除查询后面工作(在mysql中)但是peewee抛出这个错误.

(2006, "MySQL server has gone away (error(32, 'Broken pipe'))")

假设我有几个简单的模型位于food.py：

import peewee as pw

db = pw.SqliteDatabase('food.db')

class BaseModel(pw.Model):
    class Meta:
        database = db

class Taco(BaseModel):
    has_cheese = pw.BooleanField()

class Spaghetti(BaseModel):
    has_meatballs = pw.BooleanField()

db.connect()

# populate with some data if table doesn't exist
from random import random
if not Taco.table_exists():
    db.create_table(Taco)
    for _ in range(10):
        Taco.create( has_cheese = (random() < 0.5) )
    db.commit()
if not Spaghetti.table_exists():
    db.create_table(Spaghetti)
    for _ in range(10):
        Spaghetti.create( has_meatballs = (random() < 0.5) )
    db.commit()

后来，我有了food.py和food.db。但是假设Taco和 …

我有2个简单的脚本:

from sqlalchemy import create_engine, ForeignKey, Table
from sqlalchemy import Column, Date, Integer, String, DateTime, BigInteger, event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.engine import Engine
from sqlalchemy.orm import relationship, backref, sessionmaker, scoped_session, Session

class Test(declarative_base()):
    __tablename__ = "Test"
    def __init__(self, *args, **kwargs):
        args = args[0]
        for key in args:
            setattr(self, key, args[key] )
    key = Column(String, primary_key=True)

data = []
for a in range(0,10000):
    data.append({ "key" : "key%s" % a})


engine = create_engine("sqlite:///testn", echo=False)
with engine.connect() as connection:
    Test.metadata.create_all(engine)
    session …

我正在编写一组单元测试，pytest用于一些使用实现的数据库模型peewee。我想使用数据库事务（如果相关，数据库是Postgres数据库），以便在每次测试后回滚任何数据库更改。

我有一种情况，我想在测试中使用两个固定装置，但是要使两个固定装置都通过rollback方法清理其数据库模型，如下所示：

@pytest.fixture
def test_model_a():
    with db.transaction() as txn:  # `db` is my database object
        yield ModelA.create(...)
        txn.rollback()

@pytest.fixture
def test_model_b():
    with db.transaction() as txn:  # `db` is my database object
        yield ModelB.create(...)
        txn.rollback()

def test_models(test_model_a, test_model_b):
    # ...

这行得通，但是阅读文档peewee说明这很容易出错：

如果尝试使用transaction()上下文管理器与peewee嵌套事务，则仅使用最外面的事务。但是，如果嵌套块中发生异常，则可能导致无法预料的行为，因此强烈建议您使用atomic()。

但是，atomic()没有提供一种rollback()方法。似乎在显式管理事务时，关键是使用最外部的transaction()，并savepoint()在该事务内使用上下文管理器。但是可以说，在我上面的测试代码中，两个固定装置都处于同一“级别”上，我不知道在哪里创建事务以及在哪里创建保存点。

我唯一的另一个想法是使用对固定装置进行评估的顺序来决定将交易放在何处（这似乎是字母顺序的），但这确实非常脆弱。

有没有办法做到这一点？还是我的测试设计需要重新考虑？

我正在使用Peewee数据库.我有一个User3个字段的表:username,password和last_login.当用户登录到我想要更新的系统时last_login.我使用以下代码行:

from peewee import *
import datetime

class User(Model):
    username = CharField(unique=True)
    password = CharField()
    last_login = DateTimeField(default=datetime.datetime.now())

    class Meta:
        database = MySQLDatabase('mydb', user='root', charset='123456')


u=User(username="user1", last_login=datetime.datetime.now())
u.save()

虽然我没有指定任何值password,但在u.save()调用后会被覆盖.我该如何强迫小便只更新last_login字段？

如果我从带有peewee的数据库中获取多条记录，则可以将它们转换为如下所示的字典：

users = User.select().where(User.attribute == some_value).dicts()

但是，通常我只想要一条记录（或者知道只返回一条记录），所以我可以这样做：

one_user = User.get(User.name == some_value)

但是我无法调用.dicts()该对象返回的对象。
有没有办法以dict形式获取该get查询的结果？

目前，我唯一能想到的就是非Python语言

one_user = User.select().where(User.name == some_value).dicts()[0]