एक हॉबी प्रोजेक्ट के लिए मैं इस समय एक ऑब्जेक्ट ओरिएंटेड पायथन प्रोग्राम लिखना चाहता हूं। लेकिन मेरे द्वारा चलाए जा रहे पहले मुद्दे में से एक मेरे (MySQL) डेटाबेस के लिए एक क्लास बना रहा है।

मैं यथासंभव कुछ पैकेजों का उपयोग करने की कोशिश कर रहा हूं और मैं केवल pymysql का उपयोग करके डेटाबेस वर्ग को लिखने का प्रयास करना चाहता हूं। समस्या यह है कि पुस्तकालयों की भारी मात्रा है और पाइथन में एक अच्छा डेटाबेस वर्ग लिखने पर स्पष्टीकरण की कमी है।

किसी भी सुझाव और विशेष रूप से उदाहरणों की बहुत सराहना की जाएगी

3
Coen van Campenhout 24 मई 2016, 11:23
कृपया, आपके द्वारा उपयोग किए जाने वाले डेटाबेस को निर्दिष्ट करें।
 – 
dizballanze
24 मई 2016, 11:30
कोई उत्तर नहीं बल्कि एक विकल्प है: इसके बजाय sqlalchemy जैसे बड़े ORM में से किसी एक में महारत हासिल करना सीखें। अपना खुद का एक्स लिखना अभ्यास करने का एक अच्छा तरीका लगता है (कभी-कभी ऐसा होता है) लेकिन अंत में आप महत्वपूर्ण चीजों को याद कर सकते हैं या परियोजना से थक सकते हैं। यदि आप किसी अच्छी तरह से डिज़ाइन की गई और पहले से मौजूद किसी चीज़ को समझने और समझने / मास्टर करने की कोशिश करते हैं, तो आप इतने सारे किनारे के मामलों और पेचीदगियों में ठोकर खाएंगे, जिनके बारे में आप अभी सोच भी नहीं रहे हैं।
 – 
Nils Werner
24 मई 2016, 11:37
मैंने वास्तव में peewee की कोशिश की लेकिन इसका बहुत शौक नहीं था, क्या sqlalchemy एक बेहतर विकल्प है?
 – 
Coen van Campenhout
24 मई 2016, 13:18

2 जवाब

सबसे बढ़िया उत्तर

ठीक वैसी ही स्थिति होने पर, मुझे mysql-connector-python वर्ग मिला जिसके साथ मैंने अन्य वर्गों से कॉल करने के लिए एक छोटा "मॉडल" बनाया। यह एक कट-डाउन संस्करण है जो विभिन्न डीबी कॉल दिखाता है। जैसा कि आप देख सकते हैं कि मेरे पास एक कॉन्फ़िगरेशन क्लास है जिसमें सभी डीबी प्रमाणीकरण जानकारी और बहुत कुछ है)।

# dependancy: mysql-connector-python (https://dev.mysql.com/downloads/connector/python/2.1.html)
import mysql.connector
import time
import config
import HTMLParser
import StringIO

html_parser = HTMLParser.HTMLParser()

try:
    connection = mysql.connector.connect( user=config.DB_USER, password=config.DB_PASSWORD,
        host = config.DB_HOST, database=config.DB_DATABASE, unix_socket=config.UNIX_SOCKET)

    cursor = connection.cursor()

except mysql.connector.Error as err:
    logger.log('Database connection failed for '+config.DB_USER+'@'+config.DB_HOST+'/'+config.DB_DATABASE)
    exit()

def get_bad_words():
    sql = ("SELECT word FROM word_blacklist")
    results = execute(sql)
    return results

def get_moderation_method():
    sql = ("SELECT var_value FROM settings "
    "WHERE var_key = %(key)s")
    results = execute(sql, True, {'key':'moderation_method'})
    return results[0]

def current_events():
    sql = ("SELECT count(id) FROM events WHERE event_date >= DATE_SUB(NOW(), INTERVAL 2 hour) AND event_date <= DATE_ADD(NOW(), INTERVAL 5 hour)")
    results = execute(sql, True)
    return results[0]

def insert_social_post(channel, filter_type, post_id, validate, user_name, user_id, user_profile_picture, text, post_date, image_url, state):
    try:
        san_user_name = html_parser.unescape(user_name.encode('utf-8').strip()).decode("utf8").encode('ascii','ignore')
    except:
        san_user_name = html_parser.unescape(user_name.strip())
    try:
        san_text = html_parser.unescape(text.encode('utf-8').strip()).decode("utf8").encode('ascii','ignore')
    except:
        san_text = html_parser.unescape(text.strip())

    insert_post = ("INSERT IGNORE INTO social_posts "
        "(channel, filter_type, post_id, validate, user_name, user_id, user_profile_picture, text, post_date, image_url, state)"
        "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)")
    execute(insert_post, False, [channel, filter_type, str(post_id), validate,
        san_user_name.strip(), user_id, user_profile_picture, san_text.strip(), post_date, image_url, state], True)

def delete_posts(ids):
    fmt = ','.join(['%s'] * len(ids))
    cursor.execute("DELETE FROM `social_posts` WHERE id IN (%s)" % fmt,
                    tuple(ids))
    connection.commit()

def update_campaigns(campaigns):
    sql = ("UPDATE social_campaigns "
        "SET last_updated = NOW()"
        "WHERE id IN ("+(','.join(str(c) for c in campaigns))+")")
    execute(sql, False, None, True)

def execute(tuple, single = False, args = {}, commit = False):
    cursor.execute(tuple, args)

    if commit == True:
        connection.commit()
    else:
        if single == True:
            return cursor.fetchone()
        else:
            return cursor.fetchall()

def lastrowid():
    return cursor.lastrowid

def close():
    connection.close()

इस तरह कक्षा को बुलाओ:

import db
bad_words = db.get_bad_words()
3
Dharman 18 फरवरी 2020, 19:46
import sqlite3
"""singleton class to deal with db"""
'''same can be use for pymysql just replace the sqlite3 with pymysql'''


class DBConnection:
    instance = None

    def __new__(cls, *args, **kwargs):
        if cls.instance is None:
            cls.instance = super().__new__(DBConnection)
            return cls.instance
        return cls.instance

    def __init__(self, db_name='you-db-name'):
        self.name = db_name
        # connect takes url, dbname, user-id, password
        self.conn = self.connect(db_name)
        self.cursor = self.conn.cursor()

    def connect(self):
        try:
            return sqlite3.connect(self.name)
        except sqlite3.Error as e:
            pass

    def __del__(self):
        self.cursor.close()
        self.conn.close()

  # write your function here for CRUD operations
4
Dharman 18 फरवरी 2020, 19:46