How configure Posgtgres with Flet ?

[avilca]

Question

I want to use PostgresSQL as a database manager and the examples I find on the internet are with SQLite, does anyone have an example of how to configure with POSTGRESSQL with SQLAlchemy??

Quiero usar PostgresSQL como manejador de Base de Datos y y los ejemplos que encuentro en internet son con SQLite, alquien tiene algun ejemplo de como configurar con POSTGRESSQL con SQLAchemy ??

Code sample

No response

Error message

No response

------------------------------------------------------

• I have searched for answers to my question both in the issues and in previous discussions.

[dsandms]

I am using Pony ORM but it should be straight forward.

import asyncio

import flet as ft
from pony.orm import *
from datetime import datetime
import base64
import codecs
import psycopg2
from psycopg2 import sql
from psycopg2.errors import DuplicateDatabase

name = ""

def example():

db = Database()
user = #update
password = #update
host = #update
database = #update

conn = psycopg2.connect(user=user, password=password, host=host)
conn.autocommit = True
cursor = conn.cursor()

try:
    cursor.execute(psycopg2.sql.SQL('CREATE DATABASE ' + database))
except DuplicateDatabase:
    pass

db.bind(provider='postgres', user=user, password=password, host=host, database=database)
class Group(db.Entity):
    id = PrimaryKey(int, auto=True)
    group = Required(str, unique=True)
    desiredstatus = Required(bool) 
    createtime = Required(datetime,default=lambda:datetime.utcnow())
    description = Optional(str)

db.generate_mapping(create_tables=True)

listofgroups=[]
dt = ft.DataTable()

async def button_clicked(e):
    b.disabled = True
    try:
        with db_session:
            listofgroups = select(p for p in Group)
            if len(listofgroups) == 0:

                await e.control.page.update_async()
            else:
                count = 1
                dt.rows.clear()
                for group1 in listofgroups:
                    item=str(count)
                    dt.rows.insert(len(dt.rows), ft.DataRow([ft.DataCell(ft.Text(item)), ft.DataCell(ft.Text(group1.group)), ft.DataCell(ft.Text(group1.description)), ft.DataCell(ft.Text(group1.createtime))],)) 
                    await dt.update_async()
                    count += 1
    except ValueError as err1:
        dlgerr = ft.AlertDialog(title=ft.Text(err1))
        e.control.page.dialog = dlgerr
        dlgerr.open = True
        await e.control.page.update_async()

    b.disabled = False
    await e.control.page.update_async()

b = ft.ElevatedButton(text="Search", on_click=button_clicked)
dt = ft.DataTable(
        columns=[
            ft.DataColumn(ft.Text("S.No")),
            ft.DataColumn(ft.Text("Group Name")),
            ft.DataColumn(ft.Text("Group Description")),
            ft.DataColumn(ft.Text("Create On")),
        ],
    )

row0 = ft.Row([b],wrap=True, alignment=ft.MainAxisAlignment.CENTER)
row1 = ft.Row([dt],wrap=True, alignment=ft.MainAxisAlignment.CENTER)

return ft.Column(controls=[row0, row1])